From b1c80c80d1d217a758b4892060451127de63a800 Mon Sep 17 00:00:00 2001 From: Pablo Ontiveros Date: Wed, 7 Oct 2026 07:03:22 -0600 Subject: [PATCH 1/4] refactor(rebuild): inject cancellable runner --- cmd/backscroll/index_policy_test.go | 13 +- cmd/backscroll/rebuild.go | 151 ++++++++++--- cmd/backscroll/rebuild_test.go | 318 ++++++++++++++++++++++++++++ cmd/backscroll/startup_policy.go | 11 +- 4 files changed, 455 insertions(+), 38 deletions(-) create mode 100644 cmd/backscroll/rebuild_test.go diff --git a/cmd/backscroll/index_policy_test.go b/cmd/backscroll/index_policy_test.go index 0609dede..204c3d58 100644 --- a/cmd/backscroll/index_policy_test.go +++ b/cmd/backscroll/index_policy_test.go @@ -550,14 +550,15 @@ func TestValidateRobotResultLinesRejectsNonResultLines(t *testing.T) { func TestRebuildFailsOnDerivedMaintenanceError(t *testing.T) { dbPath := newSupportedIndexedConsumerDB(t) setIndexPolicyEnv(t, dbPath, t.TempDir()) - orig := rebuildBackfillDerived - rebuildBackfillDerived = func(*storage.Database, storage.BackfillDerivedOpts) error { - return fmt.Errorf("injected derived failure") - } - t.Cleanup(func() { rebuildBackfillDerived = orig }) + runner := newRebuildRunner(&rebuildDependencies{ + backfillDerivedContext: func(*storage.Database, context.Context, storage.BackfillDerivedOpts) error { + return fmt.Errorf("injected derived failure") + }, + }) var stdout, stderr bytes.Buffer - err := run(&stdout, &stderr, []string{"rebuild"}) + root := buildRootCmdWithDependencies(&stdout, &stderr, nil, nil, nil, runner) + err := runContextWithDependencies(context.Background(), &stdout, &stderr, []string{"rebuild"}, runDependencies{root: root}) if err == nil { t.Fatalf("rebuild succeeded despite derived failure; stdout=%q stderr=%q", stdout.String(), stderr.String()) } diff --git a/cmd/backscroll/rebuild.go b/cmd/backscroll/rebuild.go index 60bb6b44..8b8aeb00 100644 --- a/cmd/backscroll/rebuild.go +++ b/cmd/backscroll/rebuild.go @@ -7,12 +7,81 @@ import ( "github.com/spf13/cobra" + "github.com/pablontiv/backscroll/internal/compat" "github.com/pablontiv/backscroll/internal/config" "github.com/pablontiv/backscroll/internal/projects" "github.com/pablontiv/backscroll/internal/storage" ) -func newRebuildCmd(stdout, stderr io.Writer) *cobra.Command { +type rebuildDependencies struct { + prepareIndex func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) + closeIndexDB func(*storage.Database, error) error + rebuildFTSContext func(*storage.Database, context.Context) error + backfillDerivedContext func(*storage.Database, context.Context, storage.BackfillDerivedOpts) error + reresolveProjectsContext func(*storage.Database, context.Context, func(string) string) (int64, error) + loadRegistry func() projects.ProjectRegistry + reresolveProjectsWithRegistryContext func(*storage.Database, context.Context, projects.ProjectRegistry) (int64, error) +} + +type rebuildRunner struct { + deps rebuildDependencies +} + +func newRebuildRunner(deps *rebuildDependencies) *rebuildRunner { + resolved := rebuildDependencies{ + prepareIndex: prepareIndex, + closeIndexDB: closeIndexDB, + rebuildFTSContext: func(db *storage.Database, ctx context.Context) error { + return db.RebuildFTSContext(ctx) + }, + backfillDerivedContext: func(db *storage.Database, ctx context.Context, opts storage.BackfillDerivedOpts) error { + return db.BackfillDerivedContext(ctx, opts) + }, + reresolveProjectsContext: func(db *storage.Database, ctx context.Context, resolver func(string) string) (int64, error) { + return db.ReresolveProjectsContext(ctx, resolver) + }, + loadRegistry: projects.LoadGlobalRegistry, + reresolveProjectsWithRegistryContext: func(db *storage.Database, ctx context.Context, registry projects.ProjectRegistry) (int64, error) { + return db.ReresolveProjectsWithRegistryContext(ctx, registry) + }, + } + if deps == nil { + return &rebuildRunner{deps: resolved} + } + if deps.prepareIndex != nil { + resolved.prepareIndex = deps.prepareIndex + } + if deps.closeIndexDB != nil { + resolved.closeIndexDB = deps.closeIndexDB + } + if deps.rebuildFTSContext != nil { + resolved.rebuildFTSContext = deps.rebuildFTSContext + } + if deps.backfillDerivedContext != nil { + resolved.backfillDerivedContext = deps.backfillDerivedContext + } + if deps.reresolveProjectsContext != nil { + resolved.reresolveProjectsContext = deps.reresolveProjectsContext + } + if deps.loadRegistry != nil { + resolved.loadRegistry = deps.loadRegistry + } + if deps.reresolveProjectsWithRegistryContext != nil { + resolved.reresolveProjectsWithRegistryContext = deps.reresolveProjectsWithRegistryContext + } + return &rebuildRunner{deps: resolved} +} + +func newRebuildCmd(stdout, stderr io.Writer, runner *rebuildRunner) *cobra.Command { + if stdout == nil { + stdout = io.Discard + } + if stderr == nil { + stderr = io.Discard + } + if runner == nil { + runner = newRebuildRunner(nil) + } cmd := &cobra.Command{ Use: "rebuild", Short: "Rebuild the FTS search indexes from the database", @@ -26,49 +95,55 @@ database is the perennial event store). Use 'purge' to delete data explicitly.`, if startup.Config == nil { return fmt.Errorf("startup configuration unavailable") } - return runRebuild(cmd.Context(), stdout, stderr, startup.Config) + return runner.runRebuild(cmd.Context(), stdout, stderr, startup.Config) }, } return cmd } -var ( - rebuildBackfillDerived = func(db *storage.Database, opts storage.BackfillDerivedOpts) error { - return db.BackfillDerived(opts) +func runRebuild(ctx context.Context, stdout, stderr io.Writer, cfg *config.Config) error { + return newRebuildRunner(nil).runRebuild(ctx, stdout, stderr, cfg) +} + +func (r *rebuildRunner) runRebuild(ctx context.Context, stdout, stderr io.Writer, cfg *config.Config) (retErr error) { + if r == nil { + r = newRebuildRunner(nil) } - rebuildReresolveProjects = func(db *storage.Database, ctx context.Context, resolver func(string) string) (int64, error) { - return db.ReresolveProjects(ctx, resolver) + if stdout == nil { + stdout = io.Discard } - rebuildReresolveProjectsWithRegistry = func(db *storage.Database, ctx context.Context, registry projects.ProjectRegistry) (int64, error) { - return db.ReresolveProjectsWithRegistry(ctx, registry) + if stderr == nil { + stderr = io.Discard } -) -func runRebuild(ctx context.Context, stdout, stderr io.Writer, cfg *config.Config) (retErr error) { - db, diag, err := prepareIndex(ctx, cfg, indexMutation) + db, diag, err := r.deps.prepareIndex(ctx, cfg, indexMutation) if diag != nil { return refuseIndex(stdout, stderr, *diag, false, false) } if err != nil { return fmt.Errorf("prepare index: %w", err) } - defer func() { retErr = closeIndexDB(db, retErr) }() + defer func() { retErr = r.deps.closeIndexDB(db, retErr) }() - _, _ = fmt.Fprintf(stdout, "Re-deriving FTS indexes from database...\n") - if err := db.RebuildFTS(); err != nil { + if err := writeRebuildOutput(stdout, "Re-deriving FTS indexes from database...\n"); err != nil { + return err + } + if err := r.deps.rebuildFTSContext(db, ctx); err != nil { return fmt.Errorf("rebuild FTS: %w", err) } - // Backfill templates, corrections, and lossy tool_events for expired files - _, _ = fmt.Fprintf(stdout, "Backfilling derived data from stored text...\n") + // Backfill templates, corrections, and lossy tool_events for expired files. + if err := writeRebuildOutput(stdout, "Backfilling derived data from stored text...\n"); err != nil { + return err + } var backfillStats struct { filesProcessed int templatesFound int signalsFound int eventsExtracted int } - if err := rebuildBackfillDerived(db, storage.BackfillDerivedOpts{ + if err := r.deps.backfillDerivedContext(db, ctx, storage.BackfillDerivedOpts{ OnProgress: func(processed, templateCount, signalCount, eventCount int) { backfillStats.filesProcessed = processed backfillStats.templatesFound = templateCount @@ -79,12 +154,16 @@ func runRebuild(ctx context.Context, stdout, stderr io.Writer, cfg *config.Confi return fmt.Errorf("backfill derived: %w", err) } if backfillStats.filesProcessed > 0 { - _, _ = fmt.Fprintf(stdout, "Backfill complete: %d files processed, %d templates, %d corrections, %d lossy events.\n", - backfillStats.filesProcessed, backfillStats.templatesFound, backfillStats.signalsFound, backfillStats.eventsExtracted) + if err := writeRebuildOutput(stdout, "Backfill complete: %d files processed, %d templates, %d corrections, %d lossy events.\n", + backfillStats.filesProcessed, backfillStats.templatesFound, backfillStats.signalsFound, backfillStats.eventsExtracted); err != nil { + return err + } } - // Re-resolve project identities from session paths - _, _ = fmt.Fprintf(stdout, "Re-resolving projects from session paths...\n") + // Re-resolve project identities from session paths. + if err := writeRebuildOutput(stdout, "Re-resolving projects from session paths...\n"); err != nil { + return err + } resolver := func(sourcePath string) string { cwd := projects.DecodeCwdFromSessionPath(sourcePath) if cwd != "" { @@ -92,23 +171,35 @@ func runRebuild(ctx context.Context, stdout, stderr io.Writer, cfg *config.Confi } return "" } - resolved, err := rebuildReresolveProjects(db, ctx, resolver) + resolved, err := r.deps.reresolveProjectsContext(db, ctx, resolver) if err != nil { return fmt.Errorf("project re-resolution: %w", err) } else if resolved > 0 { - _, _ = fmt.Fprintf(stdout, "Re-resolved %d sessions with derived project identities.\n", resolved) + if err := writeRebuildOutput(stdout, "Re-resolved %d sessions with derived project identities.\n", resolved); err != nil { + return err + } } - // Registry-aware re-resolution: correct historical fallback labels - _, _ = fmt.Fprintf(stdout, "Checking registry for project label corrections...\n") - registry := projects.LoadGlobalRegistry() - registryMatched, err := rebuildReresolveProjectsWithRegistry(db, ctx, registry) + // Registry-aware re-resolution corrects historical fallback labels. + if err := writeRebuildOutput(stdout, "Checking registry for project label corrections...\n"); err != nil { + return err + } + registry := r.deps.loadRegistry() + registryMatched, err := r.deps.reresolveProjectsWithRegistryContext(db, ctx, registry) if err != nil { return fmt.Errorf("registry re-resolution: %w", err) } else if registryMatched > 0 { - _, _ = fmt.Fprintf(stdout, "Registry matched and corrected %d sessions.\n", registryMatched) + if err := writeRebuildOutput(stdout, "Registry matched and corrected %d sessions.\n", registryMatched); err != nil { + return err + } } - _, _ = fmt.Fprintf(stdout, "Rebuild complete. No indexed data was deleted (perennity contract).\n") + return writeRebuildOutput(stdout, "Rebuild complete. No indexed data was deleted (perennity contract).\n") +} + +func writeRebuildOutput(w io.Writer, format string, args ...any) error { + if _, err := fmt.Fprintf(w, format, args...); err != nil { + return fmt.Errorf("write rebuild output: %w", err) + } return nil } diff --git a/cmd/backscroll/rebuild_test.go b/cmd/backscroll/rebuild_test.go new file mode 100644 index 00000000..7f9f0e2d --- /dev/null +++ b/cmd/backscroll/rebuild_test.go @@ -0,0 +1,318 @@ +package main + +import ( + "bytes" + "context" + "errors" + "io" + "strings" + "testing" + + "github.com/pablontiv/backscroll/internal/compat" + "github.com/pablontiv/backscroll/internal/config" + "github.com/pablontiv/backscroll/internal/projects" + "github.com/pablontiv/backscroll/internal/storage" + "github.com/spf13/cobra" +) + +const ( + rebuildFTSPrefix = "Re-deriving FTS indexes from database...\n" + rebuildBackfillPrefix = "Backfilling derived data from stored text...\n" + rebuildFallbackPrefix = "Re-resolving projects from session paths...\n" + rebuildRegistryPrefix = "Checking registry for project label corrections...\n" + rebuildCompleteLine = "Rebuild complete. No indexed data was deleted (perennity contract).\n" +) + +func TestRebuildRunnerCancellationPrefixes(t *testing.T) { + tests := []struct { + name string + cancelAt string + wantOutput string + wantCalls []string + }{ + {name: "FTS", cancelAt: "fts", wantOutput: rebuildFTSPrefix, wantCalls: []string{"fts"}}, + {name: "backfill", cancelAt: "backfill", wantOutput: rebuildFTSPrefix + rebuildBackfillPrefix, wantCalls: []string{"fts", "backfill"}}, + {name: "fallback", cancelAt: "fallback", wantOutput: rebuildFTSPrefix + rebuildBackfillPrefix + rebuildFallbackPrefix, wantCalls: []string{"fts", "backfill", "fallback"}}, + {name: "registry", cancelAt: "registry", wantOutput: rebuildFTSPrefix + rebuildBackfillPrefix + rebuildFallbackPrefix + rebuildRegistryPrefix, wantCalls: []string{"fts", "backfill", "fallback", "registry"}}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var calls []string + phase := func(name string) error { + calls = append(calls, name) + if name == tc.cancelAt { + cancel() + return ctx.Err() + } + return nil + } + runner := newRebuildRunner(&rebuildDependencies{ + prepareIndex: func(got context.Context, _ *config.Config, _ indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { + if got != ctx { + t.Fatal("prepare did not receive command context") + } + return nil, nil, nil + }, + closeIndexDB: func(_ *storage.Database, err error) error { return err }, + rebuildFTSContext: func(_ *storage.Database, got context.Context) error { + if got != ctx { + t.Fatal("FTS did not receive command context") + } + return phase("fts") + }, + backfillDerivedContext: func(_ *storage.Database, got context.Context, _ storage.BackfillDerivedOpts) error { + if got != ctx { + t.Fatal("backfill did not receive command context") + } + return phase("backfill") + }, + reresolveProjectsContext: func(_ *storage.Database, got context.Context, _ func(string) string) (int64, error) { + if got != ctx { + t.Fatal("fallback did not receive command context") + } + return 0, phase("fallback") + }, + loadRegistry: func() projects.ProjectRegistry { return projects.ProjectRegistry{} }, + reresolveProjectsWithRegistryContext: func(_ *storage.Database, got context.Context, _ projects.ProjectRegistry) (int64, error) { + if got != ctx { + t.Fatal("registry did not receive command context") + } + return 0, phase("registry") + }, + }) + var stdout, stderr bytes.Buffer + err := runner.runRebuild(ctx, &stdout, &stderr, &config.Config{}) + if !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v, want context.Canceled", err) + } + if got := stdout.String(); got != tc.wantOutput { + t.Fatalf("stdout = %q, want exact prefix %q", got, tc.wantOutput) + } + if stderr.Len() != 0 { + t.Fatalf("stderr = %q, want empty", stderr.String()) + } + if strings.Contains(stdout.String(), "Rebuild complete") { + t.Fatalf("canceled output contains completion: %q", stdout.String()) + } + if strings.Join(calls, ",") != strings.Join(tc.wantCalls, ",") { + t.Fatalf("calls = %v, want %v", calls, tc.wantCalls) + } + }) + } +} + +func TestRebuildRunnerSuccessGoldenAndWriterErrors(t *testing.T) { + const golden = rebuildFTSPrefix + rebuildBackfillPrefix + + "Backfill complete: 3 files processed, 4 templates, 5 corrections, 6 lossy events.\n" + + rebuildFallbackPrefix + + "Re-resolved 7 sessions with derived project identities.\n" + + rebuildRegistryPrefix + + "Registry matched and corrected 8 sessions.\n" + + rebuildCompleteLine + + newRunner := func() *rebuildRunner { + return newRebuildRunner(&rebuildDependencies{ + prepareIndex: func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { + return nil, nil, nil + }, + closeIndexDB: func(_ *storage.Database, err error) error { return err }, + rebuildFTSContext: func(*storage.Database, context.Context) error { return nil }, + backfillDerivedContext: func(_ *storage.Database, _ context.Context, opts storage.BackfillDerivedOpts) error { + opts.OnProgress(3, 4, 5, 6) + return nil + }, + reresolveProjectsContext: func(*storage.Database, context.Context, func(string) string) (int64, error) { + return 7, nil + }, + loadRegistry: func() projects.ProjectRegistry { return projects.ProjectRegistry{} }, + reresolveProjectsWithRegistryContext: func(*storage.Database, context.Context, projects.ProjectRegistry) (int64, error) { + return 8, nil + }, + }) + } + + t.Run("success", func(t *testing.T) { + var stdout, stderr bytes.Buffer + if err := newRunner().runRebuild(context.Background(), &stdout, &stderr, &config.Config{}); err != nil { + t.Fatalf("runRebuild: %v", err) + } + if got := stdout.String(); got != golden { + t.Fatalf("stdout mismatch\n got: %q\nwant: %q", got, golden) + } + if stderr.Len() != 0 { + t.Fatalf("stderr = %q, want empty", stderr.String()) + } + }) + + for n := 0; n < len(golden); n++ { + t.Run("writer_after_"+testInt(n), func(t *testing.T) { + writeErr := errors.New("injected rebuild writer failure") + writer := &failAfterWriter{remaining: n, err: writeErr} + var stderr bytes.Buffer + err := newRunner().runRebuild(context.Background(), writer, &stderr, &config.Config{}) + if !errors.Is(err, writeErr) { + t.Fatalf("error = %v, want writer error", err) + } + if got, want := writer.String(), golden[:n]; got != want { + t.Fatalf("output = %q, want golden prefix %q", got, want) + } + if stderr.Len() != 0 { + t.Fatalf("stderr = %q, want empty", stderr.String()) + } + }) + } +} + +func TestRebuildCommandCancellationReleasesMutationLeaseExactlyOnce(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + lease := &fakeStartupLease{} + runner := newRebuildRunner(&rebuildDependencies{ + prepareIndex: func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { + return nil, nil, nil + }, + closeIndexDB: func(_ *storage.Database, err error) error { return err }, + rebuildFTSContext: func(_ *storage.Database, got context.Context) error { + if got.Done() != ctx.Done() { + t.Fatal("rebuild did not receive Cobra command cancellation") + } + cancel() + return got.Err() + }, + }) + var stdout, stderr bytes.Buffer + root := &cobra.Command{Use: "root"} + cmd := newRebuildCmd(&stdout, &stderr, runner) + registerStartupCommand(root, startupMutation, cmd) + cmd.SetContext(context.WithValue(ctx, startupContextKey{}, startupResult{Config: &config.Config{}, Lease: lease})) + + err := cmd.RunE(cmd, nil) + if !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v, want context.Canceled", err) + } + if lease.releases != 1 { + t.Fatalf("lease releases = %d, want exactly 1", lease.releases) + } + if got := stdout.String(); got != rebuildFTSPrefix { + t.Fatalf("stdout = %q, want %q", got, rebuildFTSPrefix) + } + if stderr.Len() != 0 { + t.Fatalf("stderr = %q, want empty", stderr.String()) + } +} + +func TestRebuildConstructorsAreNilSafe(t *testing.T) { + runner := newRebuildRunner(nil) + if runner == nil || runner.deps.prepareIndex == nil || runner.deps.closeIndexDB == nil || + runner.deps.rebuildFTSContext == nil || runner.deps.backfillDerivedContext == nil || + runner.deps.reresolveProjectsContext == nil || runner.deps.loadRegistry == nil || + runner.deps.reresolveProjectsWithRegistryContext == nil { + t.Fatalf("newRebuildRunner(nil) left a nil dependency: %+v", runner) + } + if cmd := newRebuildCmd(nil, nil, nil); cmd == nil || cmd.RunE == nil { + t.Fatal("newRebuildCmd with nil arguments did not return a runnable command") + } +} + +func TestRebuildRunnersParallelIsolation(t *testing.T) { + type result struct { + name string + stdout string + err error + } + started := make(chan string, 2) + release := make(chan struct{}) + results := make(chan result, 2) + + start := func(name string, fallback, registry int64) { + runner := newRebuildRunner(&rebuildDependencies{ + prepareIndex: func(context.Context, *config.Config, indexCommandClass) (*storage.Database, *compat.Diagnostic, error) { + return nil, nil, nil + }, + closeIndexDB: func(_ *storage.Database, err error) error { return err }, + rebuildFTSContext: func(*storage.Database, context.Context) error { + started <- name + <-release + return nil + }, + backfillDerivedContext: func(*storage.Database, context.Context, storage.BackfillDerivedOpts) error { return nil }, + reresolveProjectsContext: func(*storage.Database, context.Context, func(string) string) (int64, error) { + return fallback, nil + }, + loadRegistry: func() projects.ProjectRegistry { + return projects.ProjectRegistry{Projects: []projects.ProjectConfig{{ID: name}}} + }, + reresolveProjectsWithRegistryContext: func(_ *storage.Database, _ context.Context, got projects.ProjectRegistry) (int64, error) { + if len(got.Projects) != 1 || got.Projects[0].ID != name { + return 0, errors.New("runner received another runner's registry") + } + return registry, nil + }, + }) + go func() { + var stdout bytes.Buffer + err := runner.runRebuild(context.Background(), &stdout, io.Discard, &config.Config{}) + results <- result{name: name, stdout: stdout.String(), err: err} + }() + } + + start("alpha", 11, 12) + start("beta", 21, 22) + seen := map[string]bool{<-started: true, <-started: true} + if !seen["alpha"] || !seen["beta"] { + t.Fatalf("started runners = %v, want alpha and beta", seen) + } + close(release) + + wants := map[string]string{ + "alpha": rebuildFTSPrefix + rebuildBackfillPrefix + rebuildFallbackPrefix + + "Re-resolved 11 sessions with derived project identities.\n" + rebuildRegistryPrefix + + "Registry matched and corrected 12 sessions.\n" + rebuildCompleteLine, + "beta": rebuildFTSPrefix + rebuildBackfillPrefix + rebuildFallbackPrefix + + "Re-resolved 21 sessions with derived project identities.\n" + rebuildRegistryPrefix + + "Registry matched and corrected 22 sessions.\n" + rebuildCompleteLine, + } + for range 2 { + got := <-results + if got.err != nil { + t.Fatalf("%s runner: %v", got.name, got.err) + } + if got.stdout != wants[got.name] { + t.Fatalf("%s output = %q, want %q", got.name, got.stdout, wants[got.name]) + } + } +} + +type failAfterWriter struct { + bytes.Buffer + remaining int + err error +} + +func (w *failAfterWriter) Write(p []byte) (int, error) { + if len(p) <= w.remaining { + w.remaining -= len(p) + return w.Buffer.Write(p) + } + n := w.remaining + w.remaining = 0 + if n > 0 { + _, _ = w.Buffer.Write(p[:n]) + } + return n, w.err +} + +func testInt(n int) string { + if n == 0 { + return "0" + } + var digits [20]byte + i := len(digits) + for n > 0 { + i-- + digits[i] = byte('0' + n%10) + n /= 10 + } + return string(digits[i:]) +} diff --git a/cmd/backscroll/startup_policy.go b/cmd/backscroll/startup_policy.go index d14fb6d5..7cce4eb7 100644 --- a/cmd/backscroll/startup_policy.go +++ b/cmd/backscroll/startup_policy.go @@ -180,7 +180,7 @@ func buildRootCmdWithCoordinator(stdout, stderr io.Writer, coordinator *startupC return buildRootCmdWithDependencies(stdout, stderr, coordinator.defaultStartupPolicy, coordinator.syncService, recovery.Execute) } -func buildRootCmdWithDependencies(stdout, stderr io.Writer, policy startupPolicyFunc, syncService *startupSyncService, execute func(context.Context, recovery.Options) (recovery.Report, error)) *cobra.Command { +func buildRootCmdWithDependencies(stdout, stderr io.Writer, policy startupPolicyFunc, syncService *startupSyncService, execute func(context.Context, recovery.Options) (recovery.Report, error), rebuildRunners ...*rebuildRunner) *cobra.Command { if syncService == nil { syncService = newStartupSyncService() } @@ -190,6 +190,13 @@ func buildRootCmdWithDependencies(stdout, stderr io.Writer, policy startupPolicy if execute == nil { execute = recovery.Execute } + var rebuild *rebuildRunner + if len(rebuildRunners) > 0 { + rebuild = rebuildRunners[0] + } + if rebuild == nil { + rebuild = newRebuildRunner(nil) + } root := &cobra.Command{ Use: "backscroll", @@ -234,7 +241,7 @@ query merges both by rank position (RRF).`, registerStartupCommand(root, startupSnapshotRead, newContextCmd(stdout, stderr)) registerStartupCommand(root, startupSnapshotRead, newListCmd(stdout, stderr)) registerStartupCommand(root, startupSnapshotRead, newPatternsCmd(stdout, stderr)) - registerStartupCommand(root, startupMutation, newRebuildCmd(stdout, stderr)) + registerStartupCommand(root, startupMutation, newRebuildCmd(stdout, stderr, rebuild)) registerStartupCommand(root, startupMutation, newPurgeCmd(stdout, stderr)) registerStartupCommand(root, startupSnapshotRead, newValidateCmd(stdout, stderr)) registerStartupCommand(root, startupSnapshotRead, newStatusCmd(stdout, stderr)) From c0f54234ba478a0521ec6d01d169beaba1d72c55 Mon Sep 17 00:00:00 2001 From: Pablo Ontiveros Date: Wed, 7 Oct 2026 08:49:28 -0600 Subject: [PATCH 2/4] fix(storage): normalize completed cancellation rollback --- internal/storage/sync_context_test.go | 69 +++++++++++++++++++++++ internal/storage/sync_transaction.go | 25 +++++++- internal/storage/sync_transaction_test.go | 24 ++++++++ 3 files changed, 116 insertions(+), 2 deletions(-) diff --git a/internal/storage/sync_context_test.go b/internal/storage/sync_context_test.go index ad40e2f8..e751336c 100644 --- a/internal/storage/sync_context_test.go +++ b/internal/storage/sync_context_test.go @@ -7,6 +7,7 @@ import ( "fmt" "reflect" "sort" + "strings" "sync" "testing" @@ -120,6 +121,74 @@ func TestSyncFilesContextCancellationDuringBatchRollsBackEverything(t *testing.T } } +func TestSyncTransactionGateNormalizesSQLiteAutomaticCancellationRollback(t *testing.T) { + db, cleanup := newTestDB(t) + defer cleanup() + + if _, err := db.db.Exec(`CREATE TABLE cancellation_rollback_probe (value INTEGER NOT NULL)`); err != nil { + t.Fatalf("create rollback probe: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + setSyncCancellation(t, cancel) + tx, err := db.db.BeginTx(context.WithoutCancel(ctx), nil) + if err != nil { + t.Fatalf("begin transaction: %v", err) + } + defer func() { _ = tx.Rollback() }() + gate := newSyncTransactionGate(ctx, tx) + + // Hold the gate until ExecContext returns. The SQLite interrupt therefore + // completes its automatic rollback before the gate calls Rollback. + gate.mu.Lock() + _, writeErr := tx.ExecContext(ctx, fmt.Sprintf(` + WITH RECURSIVE seq(x) AS ( + VALUES(1) + UNION ALL + SELECT x + 1 FROM seq WHERE x < 100000 + ) + INSERT INTO cancellation_rollback_probe(value) + SELECT CASE WHEN x = 1000 THEN %s() ELSE x END FROM seq + `, cancelSyncFunction)) + ctxErr := ctx.Err() + if ctxErr != nil { + <-gate.callbackStarted + } + gate.mu.Unlock() + if ctxErr != context.Canceled { + t.Fatalf("context error = %v, want context.Canceled; write error = %v", ctxErr, writeErr) + } + <-gate.callbackDone + + gate.mu.Lock() + rawRollbackErr := gate.rollbackErr + gate.mu.Unlock() + var sqliteErr *sqlite.Error + if !errors.As(rawRollbackErr, &sqliteErr) { + t.Fatalf("raw rollback error = %T %v, want *sqlite.Error", rawRollbackErr, rawRollbackErr) + } + const completedRollbackMessage = "SQL logic error: cannot rollback - no transaction is active (1)" + if sqliteErr.Code() != 1 || sqliteErr.Error() != completedRollbackMessage { + t.Fatalf("raw rollback error = code %d, %q; want code 1, %q", sqliteErr.Code(), sqliteErr.Error(), completedRollbackMessage) + } + if rollbackErr := gate.rollback(); rollbackErr != nil { + t.Fatalf("normalized gate rollback = %v, want nil", rollbackErr) + } + + finalErr := writeErr + if cancelErr := gate.cancellationBeforeCommit(); cancelErr != nil && !errors.Is(finalErr, cancelErr) { + finalErr = errors.Join(cancelErr, finalErr) + } + if !errors.Is(finalErr, context.Canceled) { + t.Fatalf("final error = %v, want context.Canceled", finalErr) + } + if strings.Contains(finalErr.Error(), "rollback transaction") { + t.Fatalf("final error contains benign rollback failure: %v", finalErr) + } + + assertTableCount(t, db, "cancellation_rollback_probe", 0) +} + func TestSyncFilesContextStopwordFailureRollsBackSyncAndStopwords(t *testing.T) { db, cleanup := newTestDB(t) defer cleanup() diff --git a/internal/storage/sync_transaction.go b/internal/storage/sync_transaction.go index f3712744..38dbf592 100644 --- a/internal/storage/sync_transaction.go +++ b/internal/storage/sync_transaction.go @@ -2,9 +2,13 @@ package storage import ( "context" + "database/sql" "errors" "fmt" "sync" + + "modernc.org/sqlite" + sqlite3 "modernc.org/sqlite/lib" ) type syncTransactionFinalizer interface { @@ -146,15 +150,32 @@ func (g *syncTransactionGate) rollback() error { <-g.callbackDone } g.mu.Lock() - err := g.rollbackErr + err := normalizeSyncRollbackError(g.rollbackErr) g.mu.Unlock() return err } +func completedSyncRollbackError(err error) bool { + if errors.Is(err, sql.ErrTxDone) { + return true + } + var sqliteErr *sqlite.Error + return errors.As(err, &sqliteErr) && + sqliteErr.Code() == sqlite3.SQLITE_ERROR && + sqliteErr.Error() == "SQL logic error: cannot rollback - no transaction is active (1)" +} + +func normalizeSyncRollbackError(err error) error { + if err == nil || completedSyncRollbackError(err) { + return nil + } + return err +} + func (g *syncTransactionGate) cancellationError() error { g.mu.Lock() ctxErr := g.ctx.Err() - rollbackErr := g.rollbackErr + rollbackErr := normalizeSyncRollbackError(g.rollbackErr) g.mu.Unlock() if ctxErr == nil { ctxErr = context.Canceled diff --git a/internal/storage/sync_transaction_test.go b/internal/storage/sync_transaction_test.go index 54f5bac8..45c78db1 100644 --- a/internal/storage/sync_transaction_test.go +++ b/internal/storage/sync_transaction_test.go @@ -2,6 +2,7 @@ package storage import ( "context" + "database/sql" "errors" "sync" "testing" @@ -116,6 +117,29 @@ func TestSyncTransactionGateCancellationKeepsStableErrors(t *testing.T) { } } +func TestSyncTransactionGateRollbackNormalizesTxDone(t *testing.T) { + tx := &fakeSyncTransaction{rollbackErr: sql.ErrTxDone} + gate := newSyncTransactionGate(context.Background(), tx) + + if err := gate.rollback(); err != nil { + t.Fatalf("rollback error = %v, want nil", err) + } + if commits, rollbacks := tx.counts(); commits != 0 || rollbacks != 1 { + t.Fatalf("finalizer calls = commits %d, rollbacks %d; want 0, 1", commits, rollbacks) + } +} + +func TestSyncTransactionGateRollbackPreservesDistinctError(t *testing.T) { + rollbackFailure := errors.New("distinct rollback failure") + tx := &fakeSyncTransaction{rollbackErr: rollbackFailure} + gate := newSyncTransactionGate(context.Background(), tx) + + err := gate.rollback() + if !errors.Is(err, rollbackFailure) { + t.Fatalf("rollback error = %v, want distinct rollback failure", err) + } +} + func TestSyncTransactionGateCommitErrorIdentity(t *testing.T) { commitFailure := errors.New("commit failed") tx := &fakeSyncTransaction{commitErr: commitFailure} From 84a393ff331f2fdda6b1ef283ce6adb989c294e8 Mon Sep 17 00:00:00 2001 From: Pablo Ontiveros Date: Wed, 7 Oct 2026 09:05:34 -0600 Subject: [PATCH 3/4] fix(storage): scope rollback normalization to cancellation --- internal/storage/sync_context_test.go | 40 +++++++++++++++++++++-- internal/storage/sync_transaction.go | 8 ++--- internal/storage/sync_transaction_test.go | 20 ++++++++++-- 3 files changed, 59 insertions(+), 9 deletions(-) diff --git a/internal/storage/sync_context_test.go b/internal/storage/sync_context_test.go index e751336c..21d8a15c 100644 --- a/internal/storage/sync_context_test.go +++ b/internal/storage/sync_context_test.go @@ -12,6 +12,7 @@ import ( "testing" "modernc.org/sqlite" + sqlite3 "modernc.org/sqlite/lib" ) const cancelSyncFunction = "backscroll_test_cancel_sync_context" @@ -158,6 +159,13 @@ func TestSyncTransactionGateNormalizesSQLiteAutomaticCancellationRollback(t *tes if ctxErr != context.Canceled { t.Fatalf("context error = %v, want context.Canceled; write error = %v", ctxErr, writeErr) } + if !errors.Is(writeErr, context.Canceled) { + t.Fatalf("write error = %v, want context.Canceled", writeErr) + } + var writeSQLiteErr *sqlite.Error + if errors.As(writeErr, &writeSQLiteErr) && writeSQLiteErr.Code() != sqlite3.SQLITE_INTERRUPT { + t.Fatalf("write SQLite error code = %d, want SQLITE_INTERRUPT", writeSQLiteErr.Code()) + } <-gate.callbackDone gate.mu.Lock() @@ -168,8 +176,8 @@ func TestSyncTransactionGateNormalizesSQLiteAutomaticCancellationRollback(t *tes t.Fatalf("raw rollback error = %T %v, want *sqlite.Error", rawRollbackErr, rawRollbackErr) } const completedRollbackMessage = "SQL logic error: cannot rollback - no transaction is active (1)" - if sqliteErr.Code() != 1 || sqliteErr.Error() != completedRollbackMessage { - t.Fatalf("raw rollback error = code %d, %q; want code 1, %q", sqliteErr.Code(), sqliteErr.Error(), completedRollbackMessage) + if sqliteErr.Code() != sqlite3.SQLITE_ERROR || sqliteErr.Error() != completedRollbackMessage { + t.Fatalf("raw rollback error = code %d, %q; want SQLITE_ERROR, %q", sqliteErr.Code(), sqliteErr.Error(), completedRollbackMessage) } if rollbackErr := gate.rollback(); rollbackErr != nil { t.Fatalf("normalized gate rollback = %v, want nil", rollbackErr) @@ -186,7 +194,33 @@ func TestSyncTransactionGateNormalizesSQLiteAutomaticCancellationRollback(t *tes t.Fatalf("final error contains benign rollback failure: %v", finalErr) } - assertTableCount(t, db, "cancellation_rollback_probe", 0) + if rows := tableCount(t, db, "cancellation_rollback_probe"); rows != 0 { + t.Fatalf("interrupt path left %d rows; want canceled write, completed rollback, and zero rows", rows) + } +} + +func TestSyncTransactionGateWithoutCancellationPreservesSQLiteCompletedRollback(t *testing.T) { + db, cleanup := newTestDB(t) + defer cleanup() + + tx, err := db.db.BeginTx(context.Background(), nil) + if err != nil { + t.Fatalf("begin transaction: %v", err) + } + if _, err := tx.Exec("ROLLBACK"); err != nil { + t.Fatalf("complete transaction with SQL rollback: %v", err) + } + gate := newSyncTransactionGate(context.Background(), tx) + + err = gate.rollback() + var sqliteErr *sqlite.Error + if !errors.As(err, &sqliteErr) { + t.Fatalf("rollback error = %T %v, want *sqlite.Error", err, err) + } + const completedRollbackMessage = "SQL logic error: cannot rollback - no transaction is active (1)" + if sqliteErr.Code() != sqlite3.SQLITE_ERROR || sqliteErr.Error() != completedRollbackMessage { + t.Fatalf("rollback error = code %d, %q; want SQLITE_ERROR, %q", sqliteErr.Code(), sqliteErr.Error(), completedRollbackMessage) + } } func TestSyncFilesContextStopwordFailureRollsBackSyncAndStopwords(t *testing.T) { diff --git a/internal/storage/sync_transaction.go b/internal/storage/sync_transaction.go index 38dbf592..63e5f7ce 100644 --- a/internal/storage/sync_transaction.go +++ b/internal/storage/sync_transaction.go @@ -150,7 +150,7 @@ func (g *syncTransactionGate) rollback() error { <-g.callbackDone } g.mu.Lock() - err := normalizeSyncRollbackError(g.rollbackErr) + err := normalizeSyncRollbackError(g.rollbackErr, g.canceled) g.mu.Unlock() return err } @@ -165,8 +165,8 @@ func completedSyncRollbackError(err error) bool { sqliteErr.Error() == "SQL logic error: cannot rollback - no transaction is active (1)" } -func normalizeSyncRollbackError(err error) error { - if err == nil || completedSyncRollbackError(err) { +func normalizeSyncRollbackError(err error, canceled bool) error { + if err == nil || (canceled && completedSyncRollbackError(err)) { return nil } return err @@ -175,7 +175,7 @@ func normalizeSyncRollbackError(err error) error { func (g *syncTransactionGate) cancellationError() error { g.mu.Lock() ctxErr := g.ctx.Err() - rollbackErr := normalizeSyncRollbackError(g.rollbackErr) + rollbackErr := normalizeSyncRollbackError(g.rollbackErr, g.canceled) g.mu.Unlock() if ctxErr == nil { ctxErr = context.Canceled diff --git a/internal/storage/sync_transaction_test.go b/internal/storage/sync_transaction_test.go index 45c78db1..fb1eec8e 100644 --- a/internal/storage/sync_transaction_test.go +++ b/internal/storage/sync_transaction_test.go @@ -117,18 +117,34 @@ func TestSyncTransactionGateCancellationKeepsStableErrors(t *testing.T) { } } -func TestSyncTransactionGateRollbackNormalizesTxDone(t *testing.T) { +func TestSyncTransactionGateCancellationNormalizesTxDone(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) tx := &fakeSyncTransaction{rollbackErr: sql.ErrTxDone} - gate := newSyncTransactionGate(context.Background(), tx) + gate := newSyncTransactionGate(ctx, tx) + cancel() + <-gate.callbackDone if err := gate.rollback(); err != nil { t.Fatalf("rollback error = %v, want nil", err) } + if err := gate.cancellationError(); !errors.Is(err, context.Canceled) || errors.Is(err, sql.ErrTxDone) { + t.Fatalf("cancellation error = %v, want only context.Canceled", err) + } if commits, rollbacks := tx.counts(); commits != 0 || rollbacks != 1 { t.Fatalf("finalizer calls = commits %d, rollbacks %d; want 0, 1", commits, rollbacks) } } +func TestSyncTransactionGateWithoutCancellationPreservesTxDone(t *testing.T) { + tx := &fakeSyncTransaction{rollbackErr: sql.ErrTxDone} + gate := newSyncTransactionGate(context.Background(), tx) + + err := gate.rollback() + if !errors.Is(err, sql.ErrTxDone) { + t.Fatalf("rollback error = %v, want sql.ErrTxDone", err) + } +} + func TestSyncTransactionGateRollbackPreservesDistinctError(t *testing.T) { rollbackFailure := errors.New("distinct rollback failure") tx := &fakeSyncTransaction{rollbackErr: rollbackFailure} From 373bda84bb7fae5e76de5b486e01a8b78cd5a953 Mon Sep 17 00:00:00 2001 From: Pablo Ontiveros Date: Wed, 7 Oct 2026 09:18:36 -0600 Subject: [PATCH 4/4] docs(architecture): document routed FTS rebuild --- AGENTS.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/AGENTS.md b/AGENTS.md index b46680bd..8b430e99 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -141,7 +141,7 @@ External knowledge sources are configured with active `*.inputs.toml` manifests - **V16 parser-backed message origin**: `search_items.origin` is constrained to `human|assistant|system|automation|unknown`, with an independent `origin_version`. Readers prove origin only from native structured envelopes and tool boundaries; stored text, historical `role`, Markdown headings, and frontmatter are never inference sources. Migration defaults historical rows to `unknown`; `PendingOriginPaths` reparses only available configured sources under the shared 200-path startup cap, so expired sources remain unknown. Stable identities may enrich `unknown` to a proven origin without replacing payload or degrading prior proof, and contradictory proof is rejected. Recovery preserves matching proof, enriches compatible unknown/proven duplicates, and rejects contradictory proven origins. `patterns --kind corrections --origin ...` is opt-in, filters before pagination, and emits origin only when requested; exact reader mappings and conditional output fields are owned by `docs/input-contract.md` and `docs/patterns.md`. - **F0a rich capture (migration v8)**: readers extract per-message identity and tool metadata BEFORE serialization/cleaning destroys the evidence — `uuid` (record uuid; tool blocks get stable `#tN`/`#rN` suffixes by block index), `tool_name`, `command_head`, `is_error` (`*bool`, three-valued: tool_result blocks carry it and it is paired back onto the tool_use message cross-record via `tool_use_id`), and `was_interrupted` (detected on raw content before `CleanContent` strips "Request interrupted"). Persisted to `search_items` (`extraction_version`, `was_interrupted` columns) and the perennial `tool_events` satellite table (`UNIQUE(source_path, ordinal)`, no CASCADE lifecycle — only `purge` deletes from it, explicitly). Claude reader only; Pi/OpenCode emit zero values and stay on the legacy path. Design: `docs/superpowers/specs/2026-07-17-pattern-discovery-northstar-design.md`. - **F0a.1 command-head extraction v2 (extraction_version bump to 2)**: The `commandHead()` function in `internal/readers/claude_reader.go` now strips leading POSIX variable assignments (tokens containing `=`) before extracting the actual command name. Example: `"SP=/path/to/code; go test ./..."` extracts `"go"` instead of `"SP=/path/to/code;"`. This reduces noise in sequence mining and command pattern discovery. The bump to `extraction_version=2` triggers B1 backfill on next sync, incrementally re-extracting command_head from stored text for files indexed before this change (up to 200 files/run). -- **F0b perennial sync**: the DB is the perennial event store — session JSONL files expire (~30 days), indexed sessions survive them. Session files where EVERY message has a uuid sync append-only (no DELETE; `INSERT OR IGNORE` + UNIQUE constraints; row ids stable forever), with a one-time transition cleanup of legacy uuid-NULL rows per file. Files with any uuid-less message (Pi/OpenCode, legacy Claude) keep wipe-and-reload. `rebuild` is NON-destructive: after mutation-class startup sync has prepared the database, the handler re-derives both FTS indexes from `search_items` via FTS5 external-content `'rebuild'` in one transaction (`RebuildFTS()`) and performs no second sync — it never deletes rows and never re-reads disk as source of truth. `purge --before` is the only deletion path and deletes `tool_events` satellites explicitly in the same transaction (no CASCADE). The recover flow retains the remediation startup lock through post-install sync. +- **F0b perennial sync**: the DB is the perennial event store — session JSONL files expire (~30 days), indexed sessions survive them. Session files where EVERY message has a uuid sync append-only (no DELETE; `INSERT OR IGNORE` + UNIQUE constraints; row ids stable forever), with a one-time transition cleanup of legacy uuid-NULL rows per file. Files with any uuid-less message (Pi/OpenCode, legacy Claude) keep wipe-and-reload. `rebuild` is NON-destructive. After mutation-class startup sync prepares the database, `RebuildFTSContext()` uses one transaction to clear and repopulate `messages_fts` only with `search_items.content_type IN ('text', 'code', 'reasoning')` and `tool_fts` only with `content_type = 'tool'`. It does not use the generic FTS5 external-content `'rebuild'` command because that command would cross-index external content. It does not modify `search_items` or re-read disk. `purge --before` is the only deletion path and deletes `tool_events` satellites explicitly in the same transaction (no CASCADE). The recover flow retains the remediation startup lock through post-install sync. - **F1 exit code mining**: the **reader** parses the exit code from the FULL tool output before `toolfmt` truncates it (`SerializeToolOutput` caps at 4000 runes and a Bash exit code is usually on the last line, so parsing the capped text lost exactly the codes worth having). `ExtractExitCodeText` does the parsing with no tool gate, because a `tool_result` block lives in a different JSONL record than its `tool_use` and the tool name is not knowable at that point; the Bash gate and the code are applied when result and use are paired by `tool_use_id` — the same cross-record pairing already used for `is_error`. `SyncFiles` persists `IndexedMessage.ExitCode` verbatim and MUST NOT re-derive a code from `msg.Text`, which is truncated. The extracted exit code is stored in the `tool_events.exit_code` column (migration v8; NULL for non-Bash tools or no match). The `patterns` command aggregates tool_events by (tool_name, command_head) for commands or (tool_name, is_error, exit_code) for failures, returning top N sorted by frequency with optional filters by project, session tag, and time window. Coverage metric reports the count of events with non-NULL is_error (signalled events) against the total failure count in the result set. - **F2 template mining (migration v10)**: unsupervised Drain-inspired template miner (`internal/templates/Miner`) discovers recurring error patterns from tool output during sync. Miner uses fixed-depth token prefix clustering (depth=2) to group messages; beyond the prefix, numeric/path/UUID tokens become `<*>` variables. Error-bearing lines (is_error=true) are extracted per tool via `ExtractErrorLines` (calibrated per tool: Bash prefers LAST non-empty line + error-matching; Go test matches "--- FAIL:", "FAIL\t", "error:"; others default to error-matching heuristic) and deterministically mined with SHA256 signature. Templates stored in `message_templates` (signature, normalization_version, template_text, occurrence_count, first_seen, last_seen) joined via `template_matches` (template_id, source_path, ordinal, item_uuid) with UNIQUE constraint for idempotency. Mining runs inside `SyncFiles` transaction; re-syncing increments occurrence_count only for new matches (detected via INSERT OR IGNORE). Query method `AggregateTemplates(opts)` filters by min_support (default 3), project, date range; `patterns --kind templates [--min-support N]` exposes results in text/JSON/robot formats with normalization_version metadata. _Q1 backfill update:_ Backfill mining filters rows by: include a tool-text row only when (it has case-insensitive "error: " prefix OR its ordinal has a tool_events row with is_error=1) AND it is not an input serialization. - **One line-selection predicate for both mining paths**: `shouldMineToolLine(contentType, text, isError)` in `internal/storage/mining.go` is called by sync-time `mineTemplatesForFile` and by backfill. Only the error signal differs — sync reads `IsError` off the message, backfill derives it from an `"error: "` prefix or a `tool_events` row. The predicate exists because the two paths had drifted, and the drift was the bug: sync selected on `ToolName != ""`, which picks the **tool_use** message, whose text _is_ the input serialization, while the error text lives on the **tool_result** message that carries no `ToolName`. Sync therefore mined inputs and never errors. Tool results have no tool name, so sync mines them as `"Unknown"`, matching backfill; per-tool line calibration is a follow-up.