Skip to content

refactor(startup): inject coordinator dependencies #115

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pablontiv merged 1 commit into from
Oct 7, 2026
Merged

refactor(startup): inject coordinator dependencies #115

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 7 additions & 7 deletions cmd/backscroll/context_startup_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,23 +150,23 @@ func TestRootContextEmitsCommandDiagnosticsInEveryFormat(t *testing.T) {
}

func TestRootContextBusyFollowerUsesCommittedSnapshot(t *testing.T) {
restoreStartupCoordinatorGlobals(t)
coordinator := newStartupCoordinator()
cfg := newContextTestIndex(t)
originalPrepare := startupPrepareIndex
originalPrepare := coordinator.prepareIndex
prepareCalls := 0
startupTryAcquire = func(string) (startupLease, bool, error) { return nil, false, nil }
startupAcquire = func(context.Context, string, time.Duration) (startupLease, error) {
coordinator.tryAcquire = func(string) (startupLease, bool, error) { return nil, false, nil }
coordinator.acquire = func(context.Context, string, time.Duration) (startupLease, error) {
t.Fatal("snapshot reader must not wait for the startup lock")
return nil, errors.New("unexpected startup lock wait")
}
startupPrepareIndex = func(ctx context.Context, gotCfg *config.Config, class indexCommandClass) (*storage.Database, *compat.Diagnostic, error) {
coordinator.prepareIndex = func(ctx context.Context, gotCfg *config.Config, class indexCommandClass) (*storage.Database, *compat.Diagnostic, error) {
prepareCalls++
if class != indexDataRead {
t.Fatalf("follower prepare class=%q want %q", class, indexDataRead)
}
return originalPrepare(ctx, gotCfg, class)
}
startupSync = func(context.Context, *config.Config, io.Writer) error {
coordinator.sync = func(context.Context, *config.Config, io.Writer) error {
t.Fatal("busy snapshot follower must not synchronize")
return nil
}
Expand All @@ -175,7 +175,7 @@ func TestRootContextBusyFollowerUsesCommittedSnapshot(t *testing.T) {
startupCalls := 0
root := buildRootCmdWithStartup(&stdout, &stderr, func(ctx context.Context, progress io.Writer, class startupCommandClass) startupResult {
startupCalls++
return coordinateStartup(ctx, cfg, progress, class)
return coordinator.coordinate(ctx, cfg, progress, class)
})
root.SetArgs([]string{"context", "--uuid", "anchor", "--before", "0", "--after", "0", "--max-tokens", "16384", "--json"})

Expand Down
56 changes: 33 additions & 23 deletions cmd/backscroll/startup_coordination.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,21 +18,31 @@ const (
startupLockRetry = 50 * time.Millisecond
)

var startupMutationWait = defaultStartupMutationWait
type startupCoordinator struct {
mutationWait time.Duration
tryAcquire func(string) (startupLease, bool, error)
acquire func(context.Context, string, time.Duration) (startupLease, error)
prepareIndex func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error)
sync func(context.Context, *config.Config, io.Writer) error
}

var (
startupTryAcquire = func(path string) (startupLease, bool, error) {
return startuplock.TryAcquire(path)
}
startupAcquire = func(ctx context.Context, path string, delay time.Duration) (startupLease, error) {
return startuplock.Acquire(ctx, path, delay)
}
startupPrepareIndex = func(ctx context.Context, cfg *config.Config, class indexCommandClass) (*storage.Database, *compat.Diagnostic, error) {
return prepareIndex(ctx, cfg, class)
func newStartupCoordinator() *startupCoordinator {
return &startupCoordinator{
mutationWait: defaultStartupMutationWait,
tryAcquire: func(path string) (startupLease, bool, error) {
return startuplock.TryAcquire(path)
},
acquire: func(ctx context.Context, path string, delay time.Duration) (startupLease, error) {
return startuplock.Acquire(ctx, path, delay)
},
prepareIndex: func(ctx context.Context, cfg *config.Config, class indexCommandClass) (*storage.Database, *compat.Diagnostic, error) {
return prepareIndex(ctx, cfg, class)
},
sync: maybeAutoSyncContext,
}
)
}

func coordinateStartup(ctx context.Context, cfg *config.Config, progress io.Writer, class startupCommandClass) startupResult {
func (c *startupCoordinator) coordinate(ctx context.Context, cfg *config.Config, progress io.Writer, class startupCommandClass) startupResult {
if err := ctx.Err(); err != nil {
return canceledStartupResult(cfg, startupStageSyncLock, err)
}
Expand All @@ -43,7 +53,7 @@ func coordinateStartup(ctx context.Context, cfg *config.Config, progress io.Writ
lockStart = time.Now()
}

lease, acquired, err := startupTryAcquire(cfg.DatabasePath)
lease, acquired, err := c.tryAcquire(cfg.DatabasePath)
if ctxErr := ctx.Err(); ctxErr != nil {
result := canceledStartupResult(cfg, startupStageSyncLock, ctxErr)
if acquired {
Expand All @@ -64,12 +74,12 @@ func coordinateStartup(ctx context.Context, cfg *config.Config, progress io.Writ
return startupLockFailure(cfg, err)
}
if acquired {
return runOwnedStartup(ctx, cfg, progress, class, lease)
return c.runOwned(ctx, cfg, progress, class, lease)
}

switch class {
case startupSnapshotRead:
return concurrentSnapshotResult(ctx, cfg)
return c.concurrentSnapshotResult(ctx, cfg)
case startupMetadataRead:
return startupResult{
Config: cfg,
Expand All @@ -79,15 +89,15 @@ func coordinateStartup(ctx context.Context, cfg *config.Config, progress io.Writ
},
}
default:
waitCtx, cancel := context.WithTimeout(ctx, startupMutationWait)
waitCtx, cancel := context.WithTimeout(ctx, c.mutationWait)
defer cancel()

// Measure waiting for lock
if diagnosticsEnabled() {
lockStart = time.Now()
}

lease, err := startupAcquire(waitCtx, cfg.DatabasePath, startupLockRetry)
lease, err := c.acquire(waitCtx, cfg.DatabasePath, startupLockRetry)

// Record lock wait timing
if diagnosticsEnabled() && lockStart != (time.Time{}) {
Expand All @@ -110,11 +120,11 @@ func coordinateStartup(ctx context.Context, cfg *config.Config, progress io.Writ
}
return startupLockFailure(cfg, err)
}
return runOwnedStartup(ctx, cfg, progress, class, lease)
return c.runOwned(ctx, cfg, progress, class, lease)
}
}

func runOwnedStartup(ctx context.Context, cfg *config.Config, progress io.Writer, class startupCommandClass, lease startupLease) startupResult {
func (c *startupCoordinator) runOwned(ctx context.Context, cfg *config.Config, progress io.Writer, class startupCommandClass, lease startupLease) startupResult {
if class == startupRemediation {
return startupResult{Config: cfg, Lease: lease}
}
Expand All @@ -125,7 +135,7 @@ func runOwnedStartup(ctx context.Context, cfg *config.Config, progress io.Writer
indexPrepareStart = time.Now()
}

db, diag, err := startupPrepareIndex(ctx, cfg, indexMutation)
db, diag, err := c.prepareIndex(ctx, cfg, indexMutation)
if db != nil {
err = closeIndexDB(db, err)
}
Expand All @@ -150,7 +160,7 @@ func runOwnedStartup(ctx context.Context, cfg *config.Config, progress io.Writer
}
return ownedStartupFailureResult(cfg, class, lease, &startupFailure{Stage: startupStageIndexPrepare, Cause: err, Diagnostic: d, Recoverable: true})
}
if err := startupSync(ctx, cfg, progress); err != nil {
if err := c.sync(ctx, cfg, progress); err != nil {
activePath, _ := resolveActiveIndexPath(cfg.DatabasePath)
d := continuationFor(compat.Diagnostic{Code: compat.CodeIndexStale, Summary: fmt.Sprintf("index sync failed: %v", err)}, activePath)
return ownedStartupFailureResult(cfg, class, lease, &startupFailure{Stage: startupStageStartupSync, Cause: err, Diagnostic: d, Recoverable: true})
Expand Down Expand Up @@ -202,8 +212,8 @@ func releaseOwnedStartupResult(result startupResult, lease startupLease, stage s
return result
}

func concurrentSnapshotResult(ctx context.Context, cfg *config.Config) startupResult {
db, diag, err := startupPrepareIndex(ctx, cfg, indexDataRead)
func (c *startupCoordinator) concurrentSnapshotResult(ctx context.Context, cfg *config.Config) startupResult {
db, diag, err := c.prepareIndex(ctx, cfg, indexDataRead)
if db != nil {
err = closeIndexDB(db, err)
}
Expand Down
7 changes: 4 additions & 3 deletions cmd/backscroll/startup_coordination_process_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,15 +33,16 @@ func TestStartupCoordinationHelperProcess(t *testing.T) {
ready := os.Getenv("BACKSCROLL_SYNC_READY")
release := os.Getenv("BACKSCROLL_SYNC_RELEASE")
block := os.Getenv("BACKSCROLL_SYNC_BLOCK") == "1"
coordinator := newStartupCoordinator()
if wait := os.Getenv("BACKSCROLL_MUTATION_WAIT"); wait != "" {
parsed, err := time.ParseDuration(wait)
if err != nil {
t.Fatal(err)
}
startupMutationWait = parsed
coordinator.mutationWait = parsed
}

startupSync = func(context.Context, *config.Config, io.Writer) error {
coordinator.sync = func(context.Context, *config.Config, io.Writer) error {
if counter == "" {
return fmt.Errorf("BACKSCROLL_SYNC_COUNTER is required when startup sync runs")
}
Expand Down Expand Up @@ -88,7 +89,7 @@ func TestStartupCoordinationHelperProcess(t *testing.T) {
if err := json.Unmarshal([]byte(argvJSON), &argv); err != nil {
t.Fatal(err)
}
root := buildRootCmd(os.Stdout, os.Stderr)
root := buildRootCmdWithCoordinator(os.Stdout, os.Stderr, coordinator)
root.SetArgs(argv)
if err := root.Execute(); err != nil {
fmt.Fprintln(os.Stderr, err)
Expand Down
Loading
Loading