diff --git a/cmd/backscroll/context_startup_test.go b/cmd/backscroll/context_startup_test.go index be95645a..58db1aa0 100644 --- a/cmd/backscroll/context_startup_test.go +++ b/cmd/backscroll/context_startup_test.go @@ -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 } @@ -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"}) diff --git a/cmd/backscroll/startup_coordination.go b/cmd/backscroll/startup_coordination.go index 56fe0815..040397aa 100644 --- a/cmd/backscroll/startup_coordination.go +++ b/cmd/backscroll/startup_coordination.go @@ -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) } @@ -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 { @@ -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, @@ -79,7 +89,7 @@ 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 @@ -87,7 +97,7 @@ func coordinateStartup(ctx context.Context, cfg *config.Config, progress io.Writ 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{}) { @@ -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} } @@ -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) } @@ -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}) @@ -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) } diff --git a/cmd/backscroll/startup_coordination_process_test.go b/cmd/backscroll/startup_coordination_process_test.go index 19dad676..2b7bfe11 100644 --- a/cmd/backscroll/startup_coordination_process_test.go +++ b/cmd/backscroll/startup_coordination_process_test.go @@ -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") } @@ -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) diff --git a/cmd/backscroll/startup_coordination_test.go b/cmd/backscroll/startup_coordination_test.go index d6ee2fb5..4a2dc519 100644 --- a/cmd/backscroll/startup_coordination_test.go +++ b/cmd/backscroll/startup_coordination_test.go @@ -14,31 +14,59 @@ import ( "github.com/pablontiv/backscroll/internal/storage" ) -func restoreStartupCoordinatorGlobals(t *testing.T) { - t.Helper() - originalTryAcquire := startupTryAcquire - originalAcquire := startupAcquire - originalMutationWait := startupMutationWait - originalPrepare := startupPrepareIndex - originalSync := startupSync - t.Cleanup(func() { - startupTryAcquire = originalTryAcquire - startupAcquire = originalAcquire - startupMutationWait = originalMutationWait - startupPrepareIndex = originalPrepare - startupSync = originalSync - }) +func TestStartupCoordinatorsIsolateDependencies(t *testing.T) { + t.Parallel() + + for _, name := range []string{"first", "second"} { + name := name + t.Run(name, func(t *testing.T) { + t.Parallel() + lease := &fakeStartupLease{} + syncCalls := 0 + coordinator := &startupCoordinator{ + mutationWait: defaultStartupMutationWait, + tryAcquire: func(path string) (startupLease, bool, error) { + if !strings.Contains(path, name) { + t.Fatalf("%s coordinator received another coordinator's path %q", name, path) + } + return lease, true, nil + }, + acquire: func(context.Context, string, time.Duration) (startupLease, error) { + t.Fatal("immediate owner unexpectedly waited for lock") + return nil, nil + }, + prepareIndex: func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { + return nil, nil, nil + }, + sync: func(_ context.Context, cfg *config.Config, _ io.Writer) error { + if !strings.Contains(cfg.DatabasePath, name) { + t.Fatalf("%s sync received another coordinator's config %q", name, cfg.DatabasePath) + } + syncCalls++ + return nil + }, + } + + result := coordinator.coordinate(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), name+".db")}, io.Discard, startupSnapshotRead) + if result.Failure != nil { + t.Fatalf("startup failed: %v", result.Failure) + } + if syncCalls != 1 || lease.releases != 1 { + t.Fatalf("sync calls=%d releases=%d want 1 each", syncCalls, lease.releases) + } + }) + } } func TestCoordinateStartupImmediateOwnerSnapshotSyncsAndReleasesBeforeResult(t *testing.T) { - restoreStartupCoordinatorGlobals(t) + coordinator := newStartupCoordinator() lease := &fakeStartupLease{} syncCalls := 0 - startupTryAcquire = func(string) (startupLease, bool, error) { return lease, true, nil } - startupPrepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { + coordinator.tryAcquire = func(string) (startupLease, bool, error) { return lease, true, nil } + coordinator.prepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { return nil, nil, nil } - startupSync = func(context.Context, *config.Config, io.Writer) error { + coordinator.sync = func(context.Context, *config.Config, io.Writer) error { syncCalls++ if lease.releases != 0 { t.Fatalf("lease released before sync") @@ -46,7 +74,7 @@ func TestCoordinateStartupImmediateOwnerSnapshotSyncsAndReleasesBeforeResult(t * return nil } - result := coordinateStartup(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")}, io.Discard, startupSnapshotRead) + result := coordinator.coordinate(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")}, io.Discard, startupSnapshotRead) if result.Failure != nil { t.Fatalf("startup failed: %+v", result.Failure) } @@ -62,16 +90,16 @@ func TestCoordinateStartupImmediateOwnerSnapshotSyncsAndReleasesBeforeResult(t * } func TestCoordinateStartupImmediateOwnerMutationSyncsAndRetainsLease(t *testing.T) { - restoreStartupCoordinatorGlobals(t) + coordinator := newStartupCoordinator() lease := &fakeStartupLease{} syncCalls := 0 - startupTryAcquire = func(string) (startupLease, bool, error) { return lease, true, nil } - startupPrepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { + coordinator.tryAcquire = func(string) (startupLease, bool, error) { return lease, true, nil } + coordinator.prepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { return nil, nil, nil } - startupSync = func(context.Context, *config.Config, io.Writer) error { syncCalls++; return nil } + coordinator.sync = func(context.Context, *config.Config, io.Writer) error { syncCalls++; return nil } - result := coordinateStartup(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")}, io.Discard, startupMutation) + result := coordinator.coordinate(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")}, io.Discard, startupMutation) if result.Failure != nil { t.Fatalf("startup failed: %+v", result.Failure) } @@ -87,20 +115,20 @@ func TestCoordinateStartupImmediateOwnerMutationSyncsAndRetainsLease(t *testing. } func TestCoordinateStartupImmediateRemediationRetainsLeaseWithoutPrepareOrSync(t *testing.T) { - restoreStartupCoordinatorGlobals(t) + coordinator := newStartupCoordinator() lease := &fakeStartupLease{} - startupTryAcquire = func(string) (startupLease, bool, error) { return lease, true, nil } - startupPrepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { + coordinator.tryAcquire = func(string) (startupLease, bool, error) { return lease, true, nil } + coordinator.prepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { t.Fatal("remediation must not prepare the index") return nil, nil, nil } - startupSync = func(context.Context, *config.Config, io.Writer) error { + coordinator.sync = func(context.Context, *config.Config, io.Writer) error { t.Fatal("remediation must not run pre-handler sync") return nil } cfg := &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")} - result := coordinateStartup(context.Background(), cfg, io.Discard, startupRemediation) + result := coordinator.coordinate(context.Background(), cfg, io.Discard, startupRemediation) if result.Failure != nil { t.Fatalf("failure=%+v", result.Failure) } @@ -113,14 +141,14 @@ func TestCoordinateStartupImmediateRemediationRetainsLeaseWithoutPrepareOrSync(t } func TestCoordinateStartupBusySnapshotUsesCompatibleReadOnlySnapshot(t *testing.T) { - restoreStartupCoordinatorGlobals(t) + coordinator := newStartupCoordinator() dbPath := seedCompatibleStartupDB(t) cfg := &config.Config{DatabasePath: dbPath} syncCalls := 0 - startupTryAcquire = func(string) (startupLease, bool, error) { return nil, false, nil } - startupSync = func(context.Context, *config.Config, io.Writer) error { syncCalls++; return nil } + coordinator.tryAcquire = func(string) (startupLease, bool, error) { return nil, false, nil } + coordinator.sync = func(context.Context, *config.Config, io.Writer) error { syncCalls++; return nil } - result := coordinateStartup(context.Background(), cfg, io.Discard, startupSnapshotRead) + result := coordinator.coordinate(context.Background(), cfg, io.Discard, startupSnapshotRead) if result.Failure != nil { t.Fatalf("busy snapshot failed: %+v", result.Failure) } @@ -139,15 +167,15 @@ func TestCoordinateStartupBusySnapshotUsesCompatibleReadOnlySnapshot(t *testing. } func TestCoordinateStartupBusyMetadataSkipsReadPreparationWhenDatabaseMissing(t *testing.T) { - restoreStartupCoordinatorGlobals(t) + coordinator := newStartupCoordinator() prepareCalls := 0 - startupTryAcquire = func(string) (startupLease, bool, error) { return nil, false, nil } - startupPrepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { + coordinator.tryAcquire = func(string) (startupLease, bool, error) { return nil, false, nil } + coordinator.prepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { prepareCalls++ return nil, nil, nil } - result := coordinateStartup(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), "missing.db")}, io.Discard, startupMetadataRead) + result := coordinator.coordinate(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), "missing.db")}, io.Discard, startupMetadataRead) if result.Failure != nil { t.Fatalf("busy metadata failed: %+v", result.Failure) } @@ -160,11 +188,11 @@ func TestCoordinateStartupBusyMetadataSkipsReadPreparationWhenDatabaseMissing(t } func TestCoordinateStartupBusyMutationAcquiresWithinWaitAndBecomesOwner(t *testing.T) { - restoreStartupCoordinatorGlobals(t) + coordinator := newStartupCoordinator() lease := &fakeStartupLease{} syncCalls := 0 - startupTryAcquire = func(string) (startupLease, bool, error) { return nil, false, nil } - startupAcquire = func(ctx context.Context, path string, delay time.Duration) (startupLease, error) { + coordinator.tryAcquire = func(string) (startupLease, bool, error) { return nil, false, nil } + coordinator.acquire = func(ctx context.Context, path string, delay time.Duration) (startupLease, error) { if delay != startupLockRetry { t.Fatalf("retry delay=%v want %v", delay, startupLockRetry) } @@ -173,12 +201,12 @@ func TestCoordinateStartupBusyMutationAcquiresWithinWaitAndBecomesOwner(t *testi } return lease, nil } - startupPrepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { + coordinator.prepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { return nil, nil, nil } - startupSync = func(context.Context, *config.Config, io.Writer) error { syncCalls++; return nil } + coordinator.sync = func(context.Context, *config.Config, io.Writer) error { syncCalls++; return nil } - result := coordinateStartup(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")}, io.Discard, startupMutation) + result := coordinator.coordinate(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")}, io.Discard, startupMutation) if result.Failure != nil { t.Fatalf("startup failed: %+v", result.Failure) } @@ -191,10 +219,10 @@ func TestCoordinateStartupBusyMutationAcquiresWithinWaitAndBecomesOwner(t *testi } func TestCoordinateStartupBusyRemediationAcquiresAndBypassesPrepareSync(t *testing.T) { - restoreStartupCoordinatorGlobals(t) + coordinator := newStartupCoordinator() lease := &fakeStartupLease{} - startupTryAcquire = func(string) (startupLease, bool, error) { return nil, false, nil } - startupAcquire = func(ctx context.Context, _ string, delay time.Duration) (startupLease, error) { + coordinator.tryAcquire = func(string) (startupLease, bool, error) { return nil, false, nil } + coordinator.acquire = func(ctx context.Context, _ string, delay time.Duration) (startupLease, error) { if delay != startupLockRetry { t.Fatalf("delay=%v", delay) } @@ -203,16 +231,16 @@ func TestCoordinateStartupBusyRemediationAcquiresAndBypassesPrepareSync(t *testi } return lease, nil } - startupPrepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { + coordinator.prepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { t.Fatal("remediation must not prepare") return nil, nil, nil } - startupSync = func(context.Context, *config.Config, io.Writer) error { + coordinator.sync = func(context.Context, *config.Config, io.Writer) error { t.Fatal("remediation must not sync") return nil } - result := coordinateStartup(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")}, io.Discard, startupRemediation) + result := coordinator.coordinate(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")}, io.Discard, startupRemediation) if result.Failure != nil || result.Lease != lease { t.Fatalf("result=%+v", result) } @@ -224,23 +252,23 @@ func TestCoordinateStartupBusyRemediationAcquiresAndBypassesPrepareSync(t *testi func TestCoordinateStartupBusyMutationDeadlineReturnsSyncInProgressWithoutContinuation(t *testing.T) { for _, class := range []startupCommandClass{startupMutation, startupRemediation} { t.Run(string(class), func(t *testing.T) { - restoreStartupCoordinatorGlobals(t) - startupMutationWait = time.Millisecond - startupTryAcquire = func(string) (startupLease, bool, error) { return nil, false, nil } - startupAcquire = func(ctx context.Context, _ string, _ time.Duration) (startupLease, error) { + coordinator := newStartupCoordinator() + coordinator.mutationWait = 0 + coordinator.tryAcquire = func(string) (startupLease, bool, error) { return nil, false, nil } + coordinator.acquire = func(ctx context.Context, _ string, _ time.Duration) (startupLease, error) { <-ctx.Done() return nil, ctx.Err() } - startupPrepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { + coordinator.prepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { t.Fatal("prepare after timeout") return nil, nil, nil } - startupSync = func(context.Context, *config.Config, io.Writer) error { + coordinator.sync = func(context.Context, *config.Config, io.Writer) error { t.Fatal("sync after timeout") return nil } - result := coordinateStartup(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")}, io.Discard, class) + result := coordinator.coordinate(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")}, io.Discard, class) failure := result.startupFailure() if failure == nil || failure.Stage != startupStageSyncLock || failure.Diagnostic.Code != compat.CodeSyncInProgress || !strings.Contains(failure.Diagnostic.Summary, "retry the command") { t.Fatalf("failure=%+v", failure) @@ -253,11 +281,11 @@ func TestCoordinateStartupBusyMutationDeadlineReturnsSyncInProgressWithoutContin } func TestCoordinateStartupLockIOErrorIsSyncLockFailureNotContention(t *testing.T) { - restoreStartupCoordinatorGlobals(t) + coordinator := newStartupCoordinator() lockErr := errors.New("lock sidecar I/O failed") - startupTryAcquire = func(string) (startupLease, bool, error) { return nil, false, lockErr } + coordinator.tryAcquire = func(string) (startupLease, bool, error) { return nil, false, lockErr } - result := coordinateStartup(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")}, io.Discard, startupSnapshotRead) + result := coordinator.coordinate(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")}, io.Discard, startupSnapshotRead) failure := result.startupFailure() if failure == nil { t.Fatal("lock I/O error unexpectedly succeeded") @@ -286,20 +314,20 @@ func TestCoordinateStartupOwnerFailureLeaseRetentionByCommandClass(t *testing.T) {name: "mutation_sync_failure_retains", class: startupMutation, wantRetained: true}, } { t.Run(tc.name, func(t *testing.T) { - restoreStartupCoordinatorGlobals(t) + coordinator := newStartupCoordinator() lease := &fakeStartupLease{} - startupTryAcquire = func(string) (startupLease, bool, error) { return lease, true, nil } + coordinator.tryAcquire = func(string) (startupLease, bool, error) { return lease, true, nil } prepareErr := errors.New("prepare failed") syncErr := errors.New("sync failed") - startupPrepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { + coordinator.prepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { if tc.failPrepare { return nil, nil, prepareErr } return nil, nil, nil } - startupSync = func(context.Context, *config.Config, io.Writer) error { return syncErr } + coordinator.sync = func(context.Context, *config.Config, io.Writer) error { return syncErr } - result := coordinateStartup(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")}, io.Discard, tc.class) + result := coordinator.coordinate(context.Background(), &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")}, io.Discard, tc.class) failure := result.startupFailure() if failure == nil || !failure.Recoverable { t.Fatalf("failure=%+v want recoverable startup failure", failure) diff --git a/cmd/backscroll/startup_policy.go b/cmd/backscroll/startup_policy.go index 076893d3..803af30d 100644 --- a/cmd/backscroll/startup_policy.go +++ b/cmd/backscroll/startup_policy.go @@ -106,14 +106,12 @@ func (r startupResult) release(retErr error) error { type startupContextKey struct{} -var startupSync = maybeAutoSyncContext - func startupResultFrom(cmd *cobra.Command) startupResult { result, _ := cmd.Context().Value(startupContextKey{}).(startupResult) return result } -func defaultStartupPolicy(ctx context.Context, progress io.Writer, class startupCommandClass) startupResult { +func (c *startupCoordinator) defaultStartupPolicy(ctx context.Context, progress io.Writer, class startupCommandClass) startupResult { if err := ctx.Err(); err != nil { return startupResult{Failure: &startupFailure{Stage: startupStageInputDir, Cause: err, Diagnostic: compat.Diagnostic{Code: compat.CodeMigrationFailed, Summary: err.Error()}}} } @@ -142,7 +140,7 @@ func defaultStartupPolicy(ctx context.Context, progress io.Writer, class startup if err := ctx.Err(); err != nil { return startupResult{Config: cfg, Failure: &startupFailure{Stage: startupStageActiveManifest, Cause: err, Diagnostic: compat.Diagnostic{Code: compat.CodeMigrationFailed, Summary: err.Error()}}} } - return coordinateStartup(ctx, cfg, progress, class) + return c.coordinate(ctx, cfg, progress, class) } func validateCommandBeforeStartup(cmd *cobra.Command, args []string, positional cobra.PositionalArgs, semantic func() error) error { @@ -168,7 +166,11 @@ func validateRequiredFlagsAndGroups(cmd *cobra.Command) error { } func buildRootCmd(stdout, stderr io.Writer) *cobra.Command { - return buildRootCmdWithStartup(stdout, stderr, defaultStartupPolicy) + return buildRootCmdWithCoordinator(stdout, stderr, newStartupCoordinator()) +} + +func buildRootCmdWithCoordinator(stdout, stderr io.Writer, coordinator *startupCoordinator) *cobra.Command { + return buildRootCmdWithStartup(stdout, stderr, coordinator.defaultStartupPolicy) } func buildRootCmdWithStartup(stdout, stderr io.Writer, policy startupPolicyFunc) *cobra.Command { diff --git a/cmd/backscroll/startup_policy_test.go b/cmd/backscroll/startup_policy_test.go index 4c5eb111..5f1e3b7e 100644 --- a/cmd/backscroll/startup_policy_test.go +++ b/cmd/backscroll/startup_policy_test.go @@ -423,14 +423,13 @@ func TestDefaultStartupPolicyCallsSyncExactlyOnce(t *testing.T) { dbPath := filepath.Join(t.TempDir(), "index.db") setIndexPolicyEnv(t, dbPath, t.TempDir()) calls := 0 - originalSync := startupSync - startupSync = func(context.Context, *config.Config, io.Writer) error { + coordinator := newStartupCoordinator() + coordinator.sync = func(context.Context, *config.Config, io.Writer) error { calls++ return nil } - t.Cleanup(func() { startupSync = originalSync }) - result := defaultStartupPolicy(context.Background(), io.Discard, startupMutation) + result := coordinator.defaultStartupPolicy(context.Background(), io.Discard, startupMutation) if result.Failure != nil { t.Fatalf("startup result=%+v", result) } @@ -494,15 +493,14 @@ func TestDefaultStartupPolicyNonrecoverableStages(t *testing.T) { }, } { t.Run(tc.name, func(t *testing.T) { - originalSync := startupSync - startupSync = func(context.Context, *config.Config, io.Writer) error { + coordinator := newStartupCoordinator() + coordinator.sync = func(context.Context, *config.Config, io.Writer) error { t.Fatal("startup sync should not run after nonrecoverable startup stage") return nil } - t.Cleanup(func() { startupSync = originalSync }) tc.setup(t) - result := defaultStartupPolicy(context.Background(), io.Discard, startupMutation) + result := coordinator.defaultStartupPolicy(context.Background(), io.Discard, startupMutation) failure := result.startupFailure() if failure == nil { t.Fatal("default startup unexpectedly succeeded") @@ -562,16 +560,16 @@ func TestStartupWarningsAlwaysRenderToStderr(t *testing.T) { func TestReadOwnerLeaseReleasedBeforeHandler(t *testing.T) { dbPath := filepath.Join(t.TempDir(), "index.db") setIndexPolicyEnv(t, dbPath, t.TempDir()) - restoreStartupCoordinatorGlobals(t) + coordinator := newStartupCoordinator() lease := &fakeStartupLease{} - startupTryAcquire = func(string) (startupLease, bool, error) { return lease, true, nil } - startupPrepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { + coordinator.tryAcquire = func(string) (startupLease, bool, error) { return lease, true, nil } + coordinator.prepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { return nil, nil, nil } - startupSync = func(context.Context, *config.Config, io.Writer) error { return nil } + coordinator.sync = func(context.Context, *config.Config, io.Writer) error { return nil } var stdout, stderr bytes.Buffer - root := buildRootCmdWithStartup(&stdout, &stderr, defaultStartupPolicy) + root := buildRootCmdWithCoordinator(&stdout, &stderr, coordinator) replaceRootCommandRunE(t, root, "search", func(cmd *cobra.Command, args []string) error { if lease.releases != 1 { t.Fatalf("handler saw lease releases=%d want 1", lease.releases) diff --git a/cmd/backscroll/sync_helpers_context_test.go b/cmd/backscroll/sync_helpers_context_test.go index c6c6945f..d8862c0c 100644 --- a/cmd/backscroll/sync_helpers_context_test.go +++ b/cmd/backscroll/sync_helpers_context_test.go @@ -362,13 +362,13 @@ func TestCanceledMutationReleasesRetainedLeaseExactlyOnce(t *testing.T) { func TestOwnedStartupPreservesSuccessfulProgressBytes(t *testing.T) { type startupSyncContextKey struct{} ctx := context.WithValue(context.Background(), startupSyncContextKey{}, "startup-sync") - restoreStartupCoordinatorGlobals(t) + coordinator := newStartupCoordinator() lease := &fakeStartupLease{} - startupTryAcquire = func(string) (startupLease, bool, error) { return lease, true, nil } - startupPrepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { + coordinator.tryAcquire = func(string) (startupLease, bool, error) { return lease, true, nil } + coordinator.prepareIndex = func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { return nil, nil, nil } - startupSync = func(gotCtx context.Context, _ *config.Config, progress io.Writer) error { + coordinator.sync = func(gotCtx context.Context, _ *config.Config, progress io.Writer) error { if gotCtx.Value(startupSyncContextKey{}) != "startup-sync" { t.Fatal("startup sync did not receive coordinator context") } @@ -377,7 +377,7 @@ func TestOwnedStartupPreservesSuccessfulProgressBytes(t *testing.T) { } var progress bytes.Buffer - result := coordinateStartup(ctx, &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")}, &progress, startupSnapshotRead) + result := coordinator.coordinate(ctx, &config.Config{DatabasePath: filepath.Join(t.TempDir(), "index.db")}, &progress, startupSnapshotRead) if result.Failure != nil { t.Fatalf("failure=%v", result.Failure) }