diff --git a/cmd/periscope/archive_write_backend.go b/cmd/periscope/archive_write_backend.go index 5ce22c8cf..fc3c19d64 100644 --- a/cmd/periscope/archive_write_backend.go +++ b/cmd/periscope/archive_write_backend.go @@ -75,6 +75,10 @@ type archivePushWatchHooks struct { context.Context, *syncpkg.Engine, bool, ) (bool, error) newPGPusher func(*syncpkg.Engine) *pgPusher + newDuckDBPusher func(*syncpkg.Engine) *duckDBPusher + duckDBStartupSync func( + context.Context, *syncpkg.Engine, bool, + ) (bool, error) newUnwatchedPoller func(context.Context, unwatchedPollSyncer) unwatchedRootPoller } @@ -87,7 +91,7 @@ type unwatchedRootPoller interface { Stop() } -// newArchivePushUnwatchedPoller builds the pg watch polling owner for +// newArchivePushUnwatchedPoller builds the pg/duckdb watch polling owner for // deferred scopes. The watcher's full recovery and rename promotion defer // unavailable scopes to their polling probes, and the interval push runs a // plain SyncAll that never tombstones missed deletions, so without this owner @@ -723,53 +727,143 @@ func (b *localArchiveWriteBackend) DuckDBPushWatch( if debounce <= 0 { debounce = defaultWatchDebounce } - push := func(pctx context.Context, reason pushReason, full bool) error { - pushCfg := cfg - pushCfg.Full = full - // Watch pushes are automatic: a mirror held by a live serve - // process defers instead of rebuilding the whole archive on - // every changed batch, and archive-scale diagnostics are - // skipped. Push ignores the defer behavior when full is set. - pushCfg.Automatic = true - var res duckdbsync.PushResult - var err error - if b.watchHooks != nil && b.watchHooks.duckDBPush != nil { - res, err = b.watchHooks.duckDBPush(pctx, reason, full) - } else { - res, err = b.DuckDBPush( - pctx, duckCfg, pushCfg, projects, exclude, - ) - } - if err != nil { - return err + for _, def := range parser.Registry { + if !b.appCfg.IsUserConfigured(def.Type) { + continue } - return completeDuckDBWatchPush(res, reason) + warnMissingDirs(b.appCfg.ResolveDirs(def.Type), string(def.Type)) } + cleanResyncTemp(b.appCfg.DBPath) + + engine := syncpkg.NewEngine(b.database, syncpkg.EngineConfig{ + AgentDirs: b.appCfg.AgentDirs, + IncludeCwdPrefixes: b.appCfg.SyncIncludeCwdPrefixes, + Machine: b.appCfg.LocalMachineName, + BlockedResultCategories: b.appCfg.ResultContentBlockedCategories, + }) + defer engine.Close() + + var pusher *duckDBPusher + if b.watchHooks != nil && b.watchHooks.newDuckDBPusher != nil { + pusher = b.watchHooks.newDuckDBPusher(engine) + } else { + pusher = b.newDuckDBPusher( + func(c context.Context) error { + stats := engine.SyncAll(c, nil) + if err := c.Err(); err != nil { + return err + } + if !stats.AuthoritativeDiscoveryComplete() { + return errors.New("local sync discovery incomplete") + } + engine.FlushSignals() + return nil + }, + duckCfg, projects, exclude, + ) + } + + fmt.Printf( + "periscope duckdb watch: pushing to DuckDB "+ + "(debounce %s, floor %s)\n", + debounce, interval, + ) + loop, stopLoop := newArchivePushLoop( b.watchHooks, "duckdb watch", debounce, interval, func(c context.Context, r pushReason) error { - return push(c, r, false) + pushCfg := cfg + pushCfg.Automatic = true + if b.watchHooks != nil && b.watchHooks.duckDBPush != nil { + res, err := b.watchHooks.duckDBPush(c, r, false) + if err != nil { + return err + } + return completeDuckDBWatchPush(res, r) + } + return pusher.push(c, r, false, pushCfg) }, ) defer stopLoop() + poller := newArchivePushUnwatchedPoller(ctx, b.watchHooks, engine) + defer poller.Stop() + stopWatcher, openDispatch, unwatchedDirs := startArchivePushWatcher( - b.watchHooks, b.appCfg, nil, + b.watchHooks, b.appCfg, engine, func(callbackCtx context.Context, batch syncpkg.WatchBatch) error { + scope := func() watchRecoveryScope { + return probeWatchRecoveryScope(b.appCfg) + } + if err := syncWatchBatch(callbackCtx, engine, batch, scope); err != nil { + return err + } return notifyPushForWatchBatch(callbackCtx, loop, batch) }, - syncpkg.WatcherOptions{OnCoverageDegraded: loop.NotifyCoverageDegraded}, + syncpkg.WatcherOptions{ + OnCoverageDegraded: func(roots []string) error { + if err := poller.AddObligation(pollingObligation{ + Key: "watcher-fallback", Roots: roots, + }); err != nil { + return err + } + return loop.NotifyCoverageDegraded(roots) + }, + OnPollingRequired: func(obligation syncpkg.PollingObligation) error { + return poller.AddObligation(pollingObligation{ + Key: obligation.Key, + Roots: obligation.Roots, + Probe: obligation.Probe, + }) + }, + OnPollingReleased: poller.RemoveObligation, + }, ) defer stopWatcher() if len(unwatchedDirs) > 0 { log.Printf( - "duckdb watch: %d root(s) not watched; relying on the %s floor for coverage", - len(unwatchedDirs), interval, + "duckdb watch: %d root(s) not watched; polling every %s", + len(unwatchedDirs), unwatchedPollInterval, ) } - initialErr := push(ctx, reasonStartup, cfg.Full) + + startupSync := runPGWatchStartupSync + // DuckDB watch shares the same SyncAll-based startup path as pg watch; + // runPGWatchStartupSync is the shared default when no duckDBStartupSync + // hook overrides it (pgStartupSync is only a test harness fallback). + if b.watchHooks != nil && b.watchHooks.duckDBStartupSync != nil { + startupSync = b.watchHooks.duckDBStartupSync + } else if b.watchHooks != nil && b.watchHooks.pgStartupSync != nil { + startupSync = b.watchHooks.pgStartupSync + } + didResync, startupErr := startupSync(ctx, engine, cfg.Full) + if startupErr != nil && errors.Is(startupErr, context.Canceled) { + return nil + } + initialErr := startupErr + if initialErr == nil { + pushCfg := cfg + pushCfg.Automatic = true + if b.watchHooks != nil && b.watchHooks.duckDBPush != nil { + res, err := b.watchHooks.duckDBPush( + ctx, reasonStartup, cfg.Full || didResync, + ) + if err != nil { + initialErr = err + } else { + initialErr = completeDuckDBWatchPush(res, reasonStartup) + } + } else { + initialErr = pusher.push( + ctx, reasonStartup, cfg.Full || didResync, pushCfg, + ) + } + } if initialErr != nil { + if errors.Is(initialErr, context.Canceled) && ctx.Err() != nil { + return nil + } log.Printf("duckdb watch: initial push failed: %v", initialErr) } completePushWatchStartup(ctx, initialErr, loop, openDispatch) diff --git a/cmd/periscope/archive_write_backend_test.go b/cmd/periscope/archive_write_backend_test.go index f56041673..e26d9e76f 100644 --- a/cmd/periscope/archive_write_backend_test.go +++ b/cmd/periscope/archive_write_backend_test.go @@ -810,15 +810,21 @@ func (noopPGTarget) PushWithOptions( func (noopPGTarget) Close() error { return nil } // The watcher's full recovery and rename promotion defer unavailable scopes -// to their polling probes, and pg watch's interval push is a plain SyncAll -// that never tombstones missed deletions. Local pg watch must therefore own -// those deferred scopes with a probe-gated authoritative poller: a root the -// watcher cannot cover (here a symlinked recursive root, the same obligation -// machinery that owns roots missing at startup on portable backends) -// registers a polling obligation, and the poller reconciles its scope -// authoritatively on ticks — with no watcher event and no floor push -// involved. -func TestLocalPGPushWatchGivesDeferredScopesAPollingOwner(t *testing.T) { +// to their polling probes, and local push-watch interval pushes are a plain +// SyncAll that never tombstones missed deletions. Local pg/duckdb watch must +// therefore own those deferred scopes with a probe-gated authoritative poller. +type localPushWatchPollingOwnerSpec struct { + obligationMsg string + exitMsg string + sessionUUID string + wireHooks func(*archivePushWatchHooks) + runWatch func(context.Context, *localArchiveWriteBackend) error +} + +func testLocalPushWatchGivesDeferredScopesAPollingOwner( + t *testing.T, spec localPushWatchPollingOwnerSpec, +) { + t.Helper() dataDir := t.TempDir() dbPath := filepath.Join(dataDir, "sessions.db") database := dbtest.OpenTestDBAt(t, dbPath) @@ -856,15 +862,6 @@ func TestLocalPGPushWatchGivesDeferredScopesAPollingOwner(t *testing.T) { label: label, }, func() {} }, - pgStartupSync: func(context.Context, *syncpkg.Engine, bool) (bool, error) { - return false, nil - }, - newPGPusher: func(*syncpkg.Engine) *pgPusher { - return &pgPusher{ - localSync: func(context.Context) error { return nil }, - connect: func() (pgTarget, error) { return noopPGTarget{}, nil }, - } - }, newUnwatchedPoller: func( ctx context.Context, engine unwatchedPollSyncer, ) unwatchedRootPoller { @@ -876,41 +873,37 @@ func TestLocalPGPushWatchGivesDeferredScopesAPollingOwner(t *testing.T) { ) }, } + spec.wireHooks(backend.watchHooks) ctx, cancel := context.WithCancel(context.Background()) t.Cleanup(cancel) done := make(chan error, 1) - go func() { - done <- backend.PGPushWatch( - ctx, pgTargetSelection{}, PGPushConfig{}, nil, nil, - time.Hour, time.Hour, - ) - }() + go func() { done <- spec.runWatch(ctx, backend) }() select { case roots := <-owned: - assert.Contains(t, roots, codexRoot, - "the unwatchable root's polling obligation must reach the pg watch poller") + assert.Contains(t, roots, codexRoot, spec.obligationMsg) case err := <-done: - t.Fatalf("pg watch exited before registering obligations: %v", err) + t.Fatalf("watch exited before registering obligations: %v", err) case <-time.After(10 * time.Second): t.Fatal("no polling obligation was registered for the unwatchable root") } // A session lands under the unwatched scope; only the poller's // authoritative tick can bring it into the archive. - uuid := "d4e5f6a7-4444-4555-8666-777788889999" day := filepath.Join(codexRoot, "2026", "05", "04") require.NoError(t, os.MkdirAll(day, 0o755)) content := testjsonl.NewSessionBuilder(). AddCodexMeta( - "2026-05-04T14:00:00Z", uuid, "/home/user/code/api", + "2026-05-04T14:00:00Z", spec.sessionUUID, "/home/user/code/api", "codex_cli_rs", ). AddCodexMessage("2026-05-04T14:00:01Z", "user", "hello"). String() require.NoError(t, os.WriteFile( - filepath.Join(day, "rollout-2026-05-04T14-31-58-"+uuid+".jsonl"), + filepath.Join( + day, "rollout-2026-05-04T14-31-58-"+spec.sessionUUID+".jsonl", + ), []byte(content), 0o644, )) @@ -919,7 +912,9 @@ func TestLocalPGPushWatchGivesDeferredScopesAPollingOwner(t *testing.T) { case ticks <- time.Now(): default: } - session, err := database.GetSession(context.Background(), "codex:"+uuid) + session, err := database.GetSession( + context.Background(), "codex:"+spec.sessionUUID, + ) return err == nil && session != nil }, 10*time.Second, 20*time.Millisecond, "the returned root must be reconciled by the poller without watcher events or floor pushes") @@ -929,10 +924,78 @@ func TestLocalPGPushWatchGivesDeferredScopesAPollingOwner(t *testing.T) { case err := <-done: require.NoError(t, err) case <-time.After(30 * time.Second): - t.Fatal("pg watch did not shut down") + t.Fatal(spec.exitMsg) } } +func TestLocalPGPushWatchGivesDeferredScopesAPollingOwner(t *testing.T) { + testLocalPushWatchGivesDeferredScopesAPollingOwner( + t, localPushWatchPollingOwnerSpec{ + obligationMsg: "the unwatchable root's polling obligation must reach the pg watch poller", + exitMsg: "pg watch did not shut down", + sessionUUID: "d4e5f6a7-4444-4555-8666-777788889999", + wireHooks: func(hooks *archivePushWatchHooks) { + hooks.pgStartupSync = func( + context.Context, *syncpkg.Engine, bool, + ) (bool, error) { + return false, nil + } + hooks.newPGPusher = func(*syncpkg.Engine) *pgPusher { + return &pgPusher{ + localSync: func(context.Context) error { return nil }, + connect: func() (pgTarget, error) { + return noopPGTarget{}, nil + }, + } + } + }, + runWatch: func( + ctx context.Context, backend *localArchiveWriteBackend, + ) error { + return backend.PGPushWatch( + ctx, pgTargetSelection{}, PGPushConfig{}, nil, nil, + time.Hour, time.Hour, + ) + }, + }, + ) +} + +func TestLocalDuckDBPushWatchGivesDeferredScopesAPollingOwner(t *testing.T) { + testLocalPushWatchGivesDeferredScopesAPollingOwner( + t, localPushWatchPollingOwnerSpec{ + obligationMsg: "the unwatchable root's polling obligation must reach the duckdb watch poller", + exitMsg: "duckdb watch did not shut down", + sessionUUID: "e5f6a7b8-5555-4666-8777-888899990000", + wireHooks: func(hooks *archivePushWatchHooks) { + hooks.duckDBStartupSync = func( + context.Context, *syncpkg.Engine, bool, + ) (bool, error) { + return false, nil + } + hooks.newDuckDBPusher = func(*syncpkg.Engine) *duckDBPusher { + return &duckDBPusher{ + localSync: func(context.Context) error { return nil }, + pushMirror: func( + context.Context, DuckDBPushConfig, bool, + ) (duckdbsync.PushResult, error) { + return duckdbsync.PushResult{}, nil + }, + } + } + }, + runWatch: func( + ctx context.Context, backend *localArchiveWriteBackend, + ) error { + return backend.DuckDBPushWatch( + ctx, config.DuckDBConfig{}, DuckDBPushConfig{}, nil, nil, + time.Hour, time.Hour, + ) + }, + }, + ) +} + func testLocalArchiveWriteBackend(t *testing.T) *localArchiveWriteBackend { t.Helper() dataDir := t.TempDir() diff --git a/cmd/periscope/duckdb_watch.go b/cmd/periscope/duckdb_watch.go new file mode 100644 index 000000000..9d7276bdf --- /dev/null +++ b/cmd/periscope/duckdb_watch.go @@ -0,0 +1,72 @@ +package main + +import ( + "context" + "fmt" + + "github.com/latentsignal-org/periscope/internal/config" + duckdbsync "github.com/latentsignal-org/periscope/internal/duckdb" +) + +// duckDBPusher runs a local sync then pushes to DuckDB. The watch loop keeps +// one pusher for its lifetime so sync and push share the same sync engine +// state as the unwatched-root poller. +type duckDBPusher struct { + localSync func(context.Context) error + pushMirror func( + context.Context, DuckDBPushConfig, bool, + ) (duckdbsync.PushResult, error) +} + +func (p *duckDBPusher) push( + ctx context.Context, reason pushReason, full bool, cfg DuckDBPushConfig, +) error { + if err := p.localSync(ctx); err != nil { + return fmt.Errorf("local sync: %w", err) + } + if err := ctx.Err(); err != nil { + return err + } + res, err := p.pushMirror(ctx, cfg, full) + if err != nil { + return err + } + return completeDuckDBWatchPush(res, reason) +} + +func (b *localArchiveWriteBackend) newDuckDBPusher( + localSync func(context.Context) error, + duckCfg config.DuckDBConfig, + projects, exclude []string, +) *duckDBPusher { + return &duckDBPusher{ + localSync: localSync, + pushMirror: func( + ctx context.Context, cfg DuckDBPushConfig, forceFull bool, + ) (duckdbsync.PushResult, error) { + return b.pushDuckDBMirror( + ctx, duckCfg, cfg, projects, exclude, forceFull, + ) + }, + } +} + +func (b *localArchiveWriteBackend) pushDuckDBMirror( + ctx context.Context, + duckCfg config.DuckDBConfig, + cfg DuckDBPushConfig, + projects, excludeProjects []string, + forceFull bool, +) (duckdbsync.PushResult, error) { + if err := duckdbsync.ValidatePushTarget(duckCfg); err != nil { + return duckdbsync.PushResult{}, err + } + opts := duckdbsync.SyncOptions{ + Projects: projects, + ExcludeProjects: excludeProjects, + Automatic: cfg.Automatic, + } + return duckdbsync.Push( + ctx, duckCfg.Path, b.database, duckCfg.MachineName, opts, forceFull, nil, + ) +}