From 9b1a25c21995dc33edf8982fe18db9fc1c17aa7c Mon Sep 17 00:00:00 2001 From: Khaliq Date: Tue, 4 Aug 2026 19:51:27 +0200 Subject: [PATCH] fix(cli): unify teardown drain into a single bounded flush mode --flush-outbox-once is now the sole bounded teardown mode in both relayfile-mount and the relayfile CLI, replacing the --flush-outbox-once/--push-local-once split. File watcher observations are durably journaled before debounce under .relay/outbox/local-pending so a flush ingests only journaled paths with scan-free state saves (O(pending), never O(tree)), then force-flushes persisted outbox records. Generation-checked clearing prevents an in-flight upload from erasing a newer event. Fixes #305. Verified independently: - go build ./... - go test ./cmd/relayfile-cli/... -count=1 - go test ./cmd/relayfile-mount/... -count=1 - go test ./internal/mountsync/... -count=1 --- cmd/relayfile-cli/main.go | 15 ++- cmd/relayfile-cli/main_test.go | 117 ++++++++++++++++ cmd/relayfile-mount/main.go | 16 +-- internal/mountsync/pending_local.go | 165 +++++++++++++++++++++++ internal/mountsync/pending_local_test.go | 41 ++++++ internal/mountsync/syncer.go | 104 +++++--------- internal/mountsync/syncer_test.go | 42 +++--- internal/mountsync/watcher.go | 9 ++ internal/mountsync/watcher_test.go | 29 ++++ 9 files changed, 441 insertions(+), 97 deletions(-) create mode 100644 internal/mountsync/pending_local.go create mode 100644 internal/mountsync/pending_local_test.go diff --git a/cmd/relayfile-cli/main.go b/cmd/relayfile-cli/main.go index 9f0aa45b..af38f785 100644 --- a/cmd/relayfile-cli/main.go +++ b/cmd/relayfile-cli/main.go @@ -6410,6 +6410,7 @@ func runMount(args []string) error { logFileFlag := fs.String("log-file", "", "log file path for background mode") daemonized := fs.Bool("daemonized", false, "internal flag used by relayfile mount --background") once := fs.Bool("once", false, "run one sync cycle and exit") + flushOutboxOnce := fs.Bool("flush-outbox-once", false, "ingest watcher-journaled local drafts, flush the durable outbox, and exit without scanning or reconciling the mirror") resetAfterClobber := fs.Bool("reset-after-clobber", boolEnv("RELAYFILE_RESET_AFTER_CLOBBER", false), "acknowledge a mount-root clobber and authorize daemon to recreate the directory") rehome := fs.Bool("rehome", false, "allow re-homing an already-registered workspace mirror to a different LOCAL_DIR") if err := fs.Parse(normalizeFlagArgs(args, map[string]bool{ @@ -6439,6 +6440,7 @@ func runMount(args []string) error { "log-file": true, "daemonized": false, "once": false, + "flush-outbox-once": false, "reset-after-clobber": false, "rehome": false, "local-dir": true, @@ -6835,7 +6837,7 @@ func runMount(args []string) error { } return spawnBackgroundMountProcessFn(args, resolvedRemotePaths, absLocalDir, pidFile, logFile, resolvedLocalLayout) } - registerPID := shouldRegisterMountPID(*daemonized, *once) + registerPID := shouldRegisterMountPID(*daemonized, *once || *flushOutboxOnce) if *daemonized { if err := rotateLogFile(logFile); err != nil { return err @@ -6900,6 +6902,14 @@ func runMount(args []string) error { if initialCredExpiresAt != "" { syncer.SetCredentialExpiry(initialCredExpiresAt) } + if *flushOutboxOnce { + drainCtx, cancel := context.WithTimeout(scopeCtx, *timeout) + defer cancel() + if err := syncer.FlushOutboxOnce(drainCtx); err != nil { + return fmt.Errorf("drain writebacks for %s: %w", scope.RemotePath, err) + } + return nil + } return runMountLoopWithAuthLock( scopeCtx, syncer, @@ -7095,6 +7105,9 @@ Common flags: --interval 30s sync interval (default 30s) --background detach and keep syncing in the background --once run one sync cycle and exit (used by setup/CI) + --flush-outbox-once + bounded teardown: ingest watcher-journaled local drafts, + flush the durable outbox, and exit without a tree scan --timeout 5m per-sync timeout --bootstrap-timeout 0s hard cap for initial/full-tree bootstrap (0 = progress-based) diff --git a/cmd/relayfile-cli/main_test.go b/cmd/relayfile-cli/main_test.go index 71b4e9c2..a0e23ec6 100644 --- a/cmd/relayfile-cli/main_test.go +++ b/cmd/relayfile-cli/main_test.go @@ -2,6 +2,7 @@ package main import ( "bytes" + "context" "encoding/base64" "encoding/json" "errors" @@ -23,6 +24,7 @@ import ( "github.com/agentworkforce/relayfile/internal/delegatedauth" "github.com/agentworkforce/relayfile/internal/mountscope" "github.com/agentworkforce/relayfile/internal/mountsync" + "github.com/fsnotify/fsnotify" ) const relayfileCLITestSubprocessEnv = "RELAYFILE_CLI_TEST_SUBPROCESS" @@ -2404,6 +2406,121 @@ func TestMountUsesRecordedLocalDirWhenOmitted(t *testing.T) { } } +func TestMountFlushOutboxOnceIngestsJournaledDraftWithoutTreeScan(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + clearRelayfileEnv(t) + + const ( + workspaceID = "ws_bounded_teardown" + remoteRoot = "/slack/channels/C123/messages" + ) + localDir := t.TempDir() + stateDir := t.TempDir() + if err := os.WriteFile(filepath.Join(localDir, "unrelated.txt"), []byte("must not upload"), 0o644); err != nil { + t.Fatalf("seed unrelated file: %v", err) + } + + token := testJWTWithWorkspace(workspaceID) + var bulkCalls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + wantPath := remoteRoot + "/command.json" + switch { + case r.Method == http.MethodPost && r.URL.Path == "/v1/workspaces/"+workspaceID+"/fs/bulk": + bulkCalls.Add(1) + var req bulkWriteRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Fatalf("decode bounded drain request: %v", err) + } + if len(req.Files) != 1 { + t.Fatalf("bounded drain files = %+v, want only journaled draft", req.Files) + } + file := req.Files[0] + if file.Path != wantPath || file.Content != `{"text":"final reply"}` { + t.Fatalf("bounded drain file = %+v, want path %s and final reply", file, wantPath) + } + _, _ = io.WriteString(w, `{"written":1,"errorCount":0,"correlationId":"corr_drain","results":[{"path":"`+wantPath+`","revision":"rev_1","opId":"op_drain","writeback":{"provider":"slack","state":"succeeded"}}]}`) + case r.Method == http.MethodGet && r.URL.Path == "/v1/workspaces/"+workspaceID+"/ops/op_drain": + _, _ = io.WriteString(w, `{"opId":"op_drain","path":"`+wantPath+`","status":"succeeded","revision":"rev_1"}`) + default: + t.Fatalf("bounded drain made unexpected request: %s %s", r.Method, r.URL.Path) + } + })) + defer server.Close() + + producer, err := mountsync.NewSyncer( + mountsync.NewHTTPClient(server.URL, token, server.Client()), + mountsync.SyncerOptions{ + WorkspaceID: workspaceID, + RemoteRoot: remoteRoot, + LocalRoot: localDir, + StateDir: stateDir, + MountKind: mountsync.MountKindDaemon, + }, + ) + if err != nil { + t.Fatalf("create producer syncer: %v", err) + } + watchCtx, cancelWatch := context.WithCancel(context.Background()) + watcher, err := producer.NewFileWatcher(func(string, fsnotify.Op) {}) + if err != nil { + t.Fatalf("create producer watcher: %v", err) + } + if err := watcher.Start(watchCtx); err != nil { + t.Fatalf("start producer watcher: %v", err) + } + if err := os.WriteFile(filepath.Join(localDir, "command.json"), []byte(`{"text":"final reply"}`), 0o644); err != nil { + t.Fatalf("write final draft: %v", err) + } + journalDir := filepath.Join(localDir, ".relay", "outbox", "local-pending") + deadline := time.Now().Add(2 * time.Second) + for { + entries, readErr := os.ReadDir(journalDir) + if readErr == nil && len(entries) == 1 { + break + } + if time.Now().After(deadline) { + t.Fatalf("watcher did not persist pending draft before teardown: entries=%d err=%v", len(entries), readErr) + } + time.Sleep(5 * time.Millisecond) + } + cancelWatch() + if err := watcher.Close(); err != nil { + t.Fatalf("close producer watcher: %v", err) + } + + err = run([]string{ + "mount", workspaceID, localDir, + "--server", server.URL, + "--token", token, + "--remote-path", remoteRoot, + "--state-dir", stateDir, + "--mount-kind", mountsync.MountKindDaemon, + "--timeout", "2s", + "--flush-outbox-once", + }, strings.NewReader(""), &bytes.Buffer{}, &bytes.Buffer{}) + if err != nil { + t.Fatalf("run bounded teardown drain: %v", err) + } + if got := bulkCalls.Load(); got != 1 { + t.Fatalf("bulk calls = %d, want 1", got) + } + entries, err := os.ReadDir(journalDir) + if err != nil { + t.Fatalf("read drained journal: %v", err) + } + if len(entries) != 0 { + t.Fatalf("pending local journal still has %d entries", len(entries)) + } +} + +func TestMountRetiresPushLocalOnceSplitFlag(t *testing.T) { + err := runMount([]string{"--push-local-once"}) + if err == nil || !strings.Contains(err.Error(), "flag provided but not defined") { + t.Fatalf("retired --push-local-once returned %v", err) + } +} + func TestMountMirrorsRepeatedRemotePathsUnderScopedLayout(t *testing.T) { skipUntilScopedOperatorSurfacesReady(t) t.Setenv("HOME", t.TempDir()) diff --git a/cmd/relayfile-mount/main.go b/cmd/relayfile-mount/main.go index 9fe316f7..e6aeb2ba 100644 --- a/cmd/relayfile-mount/main.go +++ b/cmd/relayfile-mount/main.go @@ -68,7 +68,6 @@ type mountConfig struct { scopedChild bool once bool flushOutboxOnce bool - pushLocalOnce bool mode string fuseContentTTL time.Duration } @@ -112,8 +111,7 @@ func main() { fuse := flag.Bool("fuse", boolEnv("RELAYFILE_MOUNT_FUSE", false), "shortcut for --mode=fuse") fuseContentTTL := flag.Duration("fuse-content-ttl", durationEnv("RELAYFILE_MOUNT_FUSE_CONTENT_TTL", 0), "FUSE in-memory file content cache TTL (default 30s; 0 = use default)") once := flag.Bool("once", false, "run one sync cycle and exit") - flushOutboxOnce := flag.Bool("flush-outbox-once", false, "flush durable writeback outbox once and exit without reconciling the local mirror") - pushLocalOnce := flag.Bool("push-local-once", false, "ingest pending local writeback drafts (one pushLocal pass) then flush the outbox once and exit; no pullRemote/digest/reconcile — the teardown drain for last-moment drafts") + flushOutboxOnce := flag.Bool("flush-outbox-once", false, "ingest watcher-journaled local drafts, flush the durable outbox, and exit; bounded to pending paths with no mirror scan or reconcile") flag.Parse() resolvedToken := strings.TrimSpace(*token) @@ -199,7 +197,6 @@ func main() { scopes: parseTokenScopes(resolvedToken), once: *once, flushOutboxOnce: *flushOutboxOnce, - pushLocalOnce: *pushLocalOnce, mode: resolvedMode, fuseContentTTL: *fuseContentTTL, } @@ -433,22 +430,13 @@ func runSinglePollingMount(rootCtx context.Context, cfg mountConfig) error { if _, err := mountsync.StartDiagnostics(rootCtx, cfg.pprofAddr, cfg.memlogInterval, log.Default()); err != nil { return fmt.Errorf("start diagnostics: %w", err) } - if cfg.pushLocalOnce { - ctx, cancel := context.WithTimeout(rootCtx, cfg.timeout) - defer cancel() - if err := syncer.PushLocalAndFlushOnce(ctx); err != nil { - return fmt.Errorf("push local and flush once: %w", err) - } - log.Printf("local push + outbox flush completed") - return nil - } if cfg.flushOutboxOnce { ctx, cancel := context.WithTimeout(rootCtx, cfg.timeout) defer cancel() if err := syncer.FlushOutboxOnce(ctx); err != nil { return fmt.Errorf("flush outbox once: %w", err) } - log.Printf("outbox flush completed") + log.Printf("bounded writeback drain completed") return nil } log.Printf("%s", mountStartupLogLine(cfg)) diff --git a/internal/mountsync/pending_local.go b/internal/mountsync/pending_local.go new file mode 100644 index 00000000..d303f032 --- /dev/null +++ b/internal/mountsync/pending_local.go @@ -0,0 +1,165 @@ +package mountsync + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "time" + + "github.com/fsnotify/fsnotify" +) + +type pendingLocalChange struct { + RelativePath string `json:"relativePath"` + Generation string `json:"generation"` +} + +// recordPendingLocalChange durably remembers a watcher observation before its +// debounce timer starts. The record is deliberately one small file per path: +// a fresh teardown process can enumerate O(pending) work without walking the +// mounted tree, while repeated events for one path coalesce naturally. +func (s *Syncer) recordPendingLocalChange(relativePath string) error { + relativePath, err := normalizePendingLocalPath(relativePath) + if err != nil { + return err + } + record := pendingLocalChange{ + RelativePath: relativePath, + Generation: fmt.Sprintf( + "%d-%d", + time.Now().UTC().UnixNano(), + s.pendingLocalSequence.Add(1), + ), + } + payload, err := json.Marshal(record) + if err != nil { + return err + } + if err := os.MkdirAll(s.pendingLocalDir, 0o755); err != nil { + return err + } + return writeFileAtomic(s.pendingLocalChangePath(relativePath), payload, 0o644) +} + +func (s *Syncer) pendingLocalChangePath(relativePath string) string { + sum := sha256.Sum256([]byte(filepath.ToSlash(relativePath))) + return filepath.Join(s.pendingLocalDir, hex.EncodeToString(sum[:])+".json") +} + +func normalizePendingLocalPath(relativePath string) (string, error) { + cleaned := filepath.Clean(strings.TrimSpace(relativePath)) + if cleaned == "" || cleaned == "." || filepath.IsAbs(cleaned) || cleaned == ".." || strings.HasPrefix(cleaned, ".."+string(os.PathSeparator)) { + return "", fmt.Errorf("invalid pending local path %q", relativePath) + } + return filepath.ToSlash(cleaned), nil +} + +func (s *Syncer) pendingLocalChangeToken(relativePath string) []byte { + payload, err := os.ReadFile(s.pendingLocalChangePath(relativePath)) + if err != nil { + return nil + } + return payload +} + +// clearPendingLocalChange removes only the generation the handler observed. +// If another filesystem event rewrites the record while the upload is in +// flight, its new generation survives for a later callback or teardown drain. +func (s *Syncer) clearPendingLocalChange(relativePath string, observed []byte) error { + if len(observed) == 0 { + return nil + } + path := s.pendingLocalChangePath(relativePath) + current, err := os.ReadFile(path) + if errors.Is(err, os.ErrNotExist) { + return nil + } + if err != nil { + return err + } + if !bytes.Equal(current, observed) { + return nil + } + if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + return nil +} + +func (s *Syncer) pendingLocalChanges() ([]pendingLocalChange, error) { + entries, err := os.ReadDir(s.pendingLocalDir) + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + if err != nil { + return nil, err + } + records := make([]pendingLocalChange, 0, len(entries)) + var readErrors []error + for _, entry := range entries { + if entry.IsDir() || filepath.Ext(entry.Name()) != ".json" { + continue + } + path := filepath.Join(s.pendingLocalDir, entry.Name()) + payload, readErr := os.ReadFile(path) + if readErr != nil { + readErrors = append(readErrors, fmt.Errorf("read pending local change %s: %w", entry.Name(), readErr)) + continue + } + var record pendingLocalChange + if decodeErr := json.Unmarshal(payload, &record); decodeErr != nil { + readErrors = append(readErrors, fmt.Errorf("decode pending local change %s: %w", entry.Name(), decodeErr)) + continue + } + relativePath, normalizeErr := normalizePendingLocalPath(record.RelativePath) + if normalizeErr != nil { + readErrors = append(readErrors, fmt.Errorf("decode pending local change %s: %w", entry.Name(), normalizeErr)) + continue + } + record.RelativePath = relativePath + if filepath.Base(s.pendingLocalChangePath(relativePath)) != entry.Name() { + readErrors = append(readErrors, fmt.Errorf("pending local change %s does not match recorded path %q", entry.Name(), relativePath)) + continue + } + records = append(records, record) + } + sort.Slice(records, func(i, j int) bool { + return records[i].RelativePath < records[j].RelativePath + }) + return records, errors.Join(readErrors...) +} + +// FlushOutboxOnce is the single bounded teardown path. It ingests only +// paths durably observed by the watcher, then force-flushes the durable outbox. +// It never calls scanLocalFiles, pullRemote, digest generation, or websocket +// work, so its local detection cost is O(pending) rather than O(tree). +func (s *Syncer) FlushOutboxOnce(ctx context.Context) error { + if err := s.assertMountRootInvariant(); err != nil { + return err + } + records, journalErr := s.pendingLocalChanges() + errs := []error{journalErr} + for _, record := range records { + if err := ctx.Err(); err != nil { + errs = append(errs, err) + break + } + if err := s.handleLocalChange(ctx, record.RelativePath, fsnotify.Op(0), true); err != nil { + errs = append(errs, fmt.Errorf("ingest pending local change %s: %w", record.RelativePath, err)) + } + } + // Existing durable commands must still get their teardown attempt even if + // one journal entry is malformed or one local draft cannot be ingested. + if err := s.flushPersistedOutboxOnce(ctx); err != nil { + errs = append(errs, err) + } + return errors.Join(errs...) +} diff --git a/internal/mountsync/pending_local_test.go b/internal/mountsync/pending_local_test.go new file mode 100644 index 00000000..1201bc31 --- /dev/null +++ b/internal/mountsync/pending_local_test.go @@ -0,0 +1,41 @@ +package mountsync + +import ( + "os" + "testing" +) + +func TestPendingLocalJournalDoesNotClearNewerGeneration(t *testing.T) { + syncer, err := NewSyncer(&fakeClient{files: map[string]RemoteFile{}}, SyncerOptions{ + WorkspaceID: "ws_pending_generation", + RemoteRoot: "/slack/channels/C123/messages", + LocalRoot: t.TempDir(), + }) + if err != nil { + t.Fatalf("NewSyncer: %v", err) + } + const relativePath = "command.json" + if err := syncer.recordPendingLocalChange(relativePath); err != nil { + t.Fatalf("record first generation: %v", err) + } + first := syncer.pendingLocalChangeToken(relativePath) + if err := syncer.recordPendingLocalChange(relativePath); err != nil { + t.Fatalf("record newer generation: %v", err) + } + second := syncer.pendingLocalChangeToken(relativePath) + if string(first) == string(second) { + t.Fatal("successive observations reused the same journal generation") + } + if err := syncer.clearPendingLocalChange(relativePath, first); err != nil { + t.Fatalf("clear older generation: %v", err) + } + if _, err := os.Stat(syncer.pendingLocalChangePath(relativePath)); err != nil { + t.Fatalf("newer generation was cleared by older upload: %v", err) + } + if err := syncer.clearPendingLocalChange(relativePath, second); err != nil { + t.Fatalf("clear current generation: %v", err) + } + if _, err := os.Stat(syncer.pendingLocalChangePath(relativePath)); !os.IsNotExist(err) { + t.Fatalf("current generation remained after successful clear: %v", err) + } +} diff --git a/internal/mountsync/syncer.go b/internal/mountsync/syncer.go index aa25d930..1edaf50d 100644 --- a/internal/mountsync/syncer.go +++ b/internal/mountsync/syncer.go @@ -1084,12 +1084,14 @@ type Syncer struct { mountShadowDir string deadLetterDir string outboxDir string + pendingLocalDir string eventProvider string scopedChild bool scopes []string logger Logger denialLogPath string // path to .relay/permissions-denied.log state mountState + pendingLocalSequence atomic.Uint64 loaded bool bootstrapped bool // recoverStartupDrift is true only for this process instance's @@ -1611,6 +1613,7 @@ func NewSyncer(client RemoteClient, opts SyncerOptions) (*Syncer, error) { mountShadowDir := filepath.Join(localRoot, ".relay", ".mount-shadow") deadLetterDir := filepath.Join(localRoot, ".relay", "dead-letter") outboxDir := filepath.Join(localRoot, ".relay", "outbox") + pendingLocalDir := filepath.Join(outboxDir, "local-pending") scopes := normalizeScopes(opts.Scopes) if len(scopes) == 0 { if httpClient, ok := client.(*HTTPClient); ok { @@ -1644,7 +1647,7 @@ func NewSyncer(client RemoteClient, opts SyncerOptions) (*Syncer, error) { opts.Logger.Printf("quarantined %d legacy private mount state file(s) outside mounted tree", len(moved)) } } - for _, dir := range []string{outboxDir, filepath.Join(outboxDir, "pending"), filepath.Join(outboxDir, "acked"), filepath.Join(outboxDir, "failed")} { + for _, dir := range []string{outboxDir, filepath.Join(outboxDir, "pending"), filepath.Join(outboxDir, "acked"), filepath.Join(outboxDir, "failed"), pendingLocalDir} { if err := os.MkdirAll(dir, 0o755); err != nil { return nil, err } @@ -1796,6 +1799,7 @@ func NewSyncer(client RemoteClient, opts SyncerOptions) (*Syncer, error) { mountShadowDir: mountShadowDir, deadLetterDir: deadLetterDir, outboxDir: outboxDir, + pendingLocalDir: pendingLocalDir, eventProvider: eventProvider, scopedChild: opts.ScopedChild, scopes: scopes, @@ -1853,7 +1857,16 @@ func (s *Syncer) SetCredentialExpiry(expiresAt string) { // NewFileWatcher creates a watcher with the same local/remote mapping as this // Syncer so path-collision guards cannot diverge between event and scan paths. func (s *Syncer) NewFileWatcher(onChange func(string, fsnotify.Op)) (*FileWatcher, error) { - return NewFileWatcherForTopology(s.localRoot, s.remoteRoot, s.scopedChild, onChange) + watcher, err := NewFileWatcherForTopology(s.localRoot, s.remoteRoot, s.scopedChild, onChange) + if err != nil { + return nil, err + } + watcher.onObserve = func(relativePath string) { + if err := s.recordPendingLocalChange(relativePath); err != nil { + s.logf("failed to persist pending local change %s: %v", relativePath, err) + } + } + return watcher, nil } func parseScopesFromJWT(token string) []string { @@ -1932,10 +1945,10 @@ func (s *Syncer) SkipStuck(ctx context.Context, max int) (int, error) { return count, err } -// FlushOutboxOnce uploads only persisted durable outbox records and exits -// without reconciling the local mirror. It is intentionally O(outbox): no -// local tree scan, pushLocal, pullRemote, websocket, or digest work. -func (s *Syncer) FlushOutboxOnce(ctx context.Context) error { +// flushPersistedOutboxOnce uploads only persisted durable outbox records. The +// exported teardown entrypoint is FlushOutboxOnce in pending_local.go, which +// first ingests watcher-journaled local paths and then calls this helper. +func (s *Syncer) flushPersistedOutboxOnce(ctx context.Context) error { s.mu.Lock() defer s.mu.Unlock() @@ -1964,66 +1977,6 @@ func (s *Syncer) FlushOutboxOnce(ctx context.Context) error { return s.saveStateWithoutLocalScan() } -// PushLocalAndFlushOnce ingests pending local writeback drafts with a single -// pushLocal pass, then flushes the durable outbox, and exits — without -// pullRemote, digest, websocket, or a full reconcile cycle. -// -// It is the teardown drain. Local writeback drafts are normally ingested into -// the outbox by the running daemon's sync cycle (watcher + pushLocal). A draft -// written after that daemon's last cycle and just before shutdown — e.g. a -// final fire-and-forget reply right before a one-shot sandbox is torn down — is -// still on disk but not yet in the outbox, so FlushOutboxOnce (outbox-only, no -// local scan) silently drops it. Running pushLocal here, in the fresh cleanup -// process that scans the on-disk mirror, ingests those drafts before flushing. -// -// The local scan is the cost (the same O(tree) work FlushOutboxOnce exists to -// avoid), so callers should invoke this only when pending local writes are -// detected and keep FlushOutboxOnce for the no-pending-writes fast path. Unlike -// a full reconcile it still skips pullRemote/digest/websocket, so it cannot -// reintroduce the pull-side flush-124 stalls. -func (s *Syncer) PushLocalAndFlushOnce(ctx context.Context) error { - // Same top-of-cycle invariant as syncReserved: pushLocal scans and mutates - // the local mirror, so refuse to run if the mount root was wiped/clobbered - // (recovery is gated behind --reset-after-clobber). FlushOutboxOnce skips - // this because it is outbox-only and never touches the mirror. - if err := s.assertMountRootInvariant(); err != nil { - return err - } - - s.mu.Lock() - defer s.mu.Unlock() - - if err := s.loadState(); err != nil { - return err - } - conflicted, err := s.pushLocal(ctx) - if err != nil { - s.markSyncError(err) - _ = s.saveStateWithoutLocalScan() - return err - } - if err := s.flushOutboxRecords(ctx, conflicted, true); err != nil { - s.markSyncError(err) - _ = s.saveStateWithoutLocalScan() - return err - } - outbox := s.summarizeOutbox() - if outbox.NeedsAttention > 0 { - err := fmt.Errorf("outbox needs attention: %d command(s)", outbox.NeedsAttention) - s.markSyncError(err) - _ = s.saveStateWithoutLocalScan() - return err - } - if outbox.Pending > 0 { - err := fmt.Errorf("outbox pending remains: %d command(s)", outbox.Pending) - s.markSyncError(err) - _ = s.saveStateWithoutLocalScan() - return err - } - s.markSyncSuccess() - return s.saveState() -} - // HandleLocalChange routes a local filesystem event to the appropriate // writeback action. // @@ -2048,6 +2001,10 @@ func (s *Syncer) PushLocalAndFlushOnce(ctx context.Context) error { // is no actual content change, so spurious events (Chmod-only on an // unmodified file) do not generate noise on the wire. func (s *Syncer) HandleLocalChange(ctx context.Context, relativePath string, op fsnotify.Op) error { + return s.handleLocalChange(ctx, relativePath, op, false) +} + +func (s *Syncer) handleLocalChange(ctx context.Context, relativePath string, op fsnotify.Op, saveWithoutScan bool) (resultErr error) { relativePath = filepath.ToSlash(strings.TrimSpace(filepath.Clean(relativePath))) if relativePath == "" || relativePath == "." { return nil @@ -2062,6 +2019,12 @@ func (s *Syncer) HandleLocalChange(ctx context.Context, relativePath string, op } return nil } + journalToken := s.pendingLocalChangeToken(relativePath) + defer func() { + if resultErr == nil { + resultErr = s.clearPendingLocalChange(relativePath, journalToken) + } + }() s.mu.Lock() defer s.mu.Unlock() @@ -2084,10 +2047,17 @@ func (s *Syncer) HandleLocalChange(ctx context.Context, relativePath string, op saveWithStatus := func(run func() error) error { if err := run(); err != nil { s.markSyncError(err) - _ = s.saveState() + if saveWithoutScan { + _ = s.saveStateWithoutLocalScan() + } else { + _ = s.saveState() + } return err } s.markSyncSuccess() + if saveWithoutScan { + return s.saveStateWithoutLocalScan() + } return s.saveState() } diff --git a/internal/mountsync/syncer_test.go b/internal/mountsync/syncer_test.go index 35307ff6..bebffefb 100644 --- a/internal/mountsync/syncer_test.go +++ b/internal/mountsync/syncer_test.go @@ -4361,20 +4361,22 @@ func TestFlushOutboxOnceFlushesPendingWithoutMirrorScan(t *testing.T) { } } -// A draft written but never ingested by a sync cycle (the teardown race: a -// final fire-and-forget reply right before shutdown) is on disk but not in the -// outbox. FlushOutboxOnce drops it (outbox-only, no local scan); -// PushLocalAndFlushOnce ingests it by scanning the on-disk mirror, then flushes. -func TestPushLocalAndFlushOnceIngestsUnsyncedLocalDraft(t *testing.T) { +// A draft observed by the watcher but not ingested before shutdown is durably +// journaled. The fresh teardown process reads only that O(pending) journal, +// ingests the draft, and flushes it without scanning unrelated local files. +func TestFlushOutboxOnceIngestsOnlyJournaledLocalDraft(t *testing.T) { localDir := t.TempDir() if err := os.WriteFile(filepath.Join(localDir, "command.json"), []byte(`{"text":"hello"}`), 0o644); err != nil { t.Fatalf("seed local command failed: %v", err) } + if err := os.WriteFile(filepath.Join(localDir, "unrelated.txt"), []byte("must not be scanned"), 0o644); err != nil { + t.Fatalf("seed unrelated local file failed: %v", err) + } // Baseline: FlushOutboxOnce must NOT ingest the unsynced draft (the bug). flushClient := &fakeClient{files: map[string]RemoteFile{}} flushOnly, err := NewSyncer(flushClient, SyncerOptions{ - WorkspaceID: "ws_push_local_once", + WorkspaceID: "ws_bounded_drain", RemoteRoot: "/slack/channels/C123/messages", LocalRoot: localDir, }) @@ -4388,29 +4390,39 @@ func TestPushLocalAndFlushOnceIngestsUnsyncedLocalDraft(t *testing.T) { t.Fatalf("FlushOutboxOnce must not ingest an unsynced local draft, got %d uploads", flushClient.bulkWriteCalls) } - // Fix: PushLocalAndFlushOnce scans the on-disk mirror, ingests the draft, - // uploads it, and drains the outbox. + // The watcher observer writes this before its debounce callback. Teardown + // therefore knows the exact path even in a fresh process. + if err := flushOnly.recordPendingLocalChange("command.json"); err != nil { + t.Fatalf("record pending local draft: %v", err) + } + pushClient := &fakeClient{files: map[string]RemoteFile{}} drain, err := NewSyncer(pushClient, SyncerOptions{ - WorkspaceID: "ws_push_local_once", + WorkspaceID: "ws_bounded_drain", RemoteRoot: "/slack/channels/C123/messages", LocalRoot: localDir, }) if err != nil { t.Fatalf("NewSyncer drain: %v", err) } - if err := drain.PushLocalAndFlushOnce(context.Background()); err != nil { - t.Fatalf("PushLocalAndFlushOnce failed: %v", err) + if err := drain.FlushOutboxOnce(context.Background()); err != nil { + t.Fatalf("FlushOutboxOnce failed: %v", err) } if pushClient.bulkWriteCalls != 1 { t.Fatalf("expected the unsynced draft to be ingested + uploaded once, got %d", pushClient.bulkWriteCalls) } if pending := readPendingOutboxRecordsForTest(t, localDir); len(pending) != 0 { - t.Fatalf("expected outbox drained after push+flush, got %+v", pending) + t.Fatalf("expected outbox drained after bounded drain, got %+v", pending) + } + if records, err := drain.pendingLocalChanges(); err != nil || len(records) != 0 { + t.Fatalf("expected local pending journal drained, records=%+v err=%v", records, err) + } + if _, ok := pushClient.files["/slack/channels/C123/messages/unrelated.txt"]; ok { + t.Fatal("bounded drain scanned and uploaded an unjournaled local file") } } -func TestPushLocalAndFlushOnceSkipsSelfReferentialOutboxControlFiles(t *testing.T) { +func TestFlushOutboxOnceSkipsSelfReferentialOutboxControlFiles(t *testing.T) { localDir := t.TempDir() client := &fakeClient{files: map[string]RemoteFile{}} logger := &captureLogger{} @@ -4438,8 +4450,8 @@ func TestPushLocalAndFlushOnceSkipsSelfReferentialOutboxControlFiles(t *testing. t.Fatalf("seed self-referential outbox record: %v", err) } - if err := syncer.PushLocalAndFlushOnce(context.Background()); err != nil { - t.Fatalf("PushLocalAndFlushOnce failed: %v", err) + if err := syncer.FlushOutboxOnce(context.Background()); err != nil { + t.Fatalf("FlushOutboxOnce failed: %v", err) } if client.bulkWriteCalls != 0 || client.writeFileCalls != 0 { diff --git a/internal/mountsync/watcher.go b/internal/mountsync/watcher.go index 11dfc193..500885d4 100644 --- a/internal/mountsync/watcher.go +++ b/internal/mountsync/watcher.go @@ -26,6 +26,7 @@ type FileWatcher struct { remoteRoot string scopedChild bool onChange func(relativePath string, op fsnotify.Op) + onObserve func(relativePath string) maxDirs int watchedDirs int mu sync.Mutex @@ -177,6 +178,14 @@ func (fw *FileWatcher) queueChange(rel string, op fsnotify.Op) { fw.mu.Unlock() return } + // Persist observation before the debounce window starts. A teardown can + // stop the watcher before the delayed callback runs; the Syncer's observer + // leaves a durable per-path record for the fresh drain process in that + // case. This hook intentionally runs while fw.mu is held so Close cannot + // return between observing the event and recording it. + if fw.onObserve != nil { + fw.onObserve(rel) + } if t, ok := fw.debounce[rel]; ok { if t.Stop() { fw.wg.Done() diff --git a/internal/mountsync/watcher_test.go b/internal/mountsync/watcher_test.go index 9ba7e980..17d2a731 100644 --- a/internal/mountsync/watcher_test.go +++ b/internal/mountsync/watcher_test.go @@ -170,6 +170,35 @@ func TestWatcherCloseCancelsPendingDebounce(t *testing.T) { assertNoWatcherEvents(t, events, 150*time.Millisecond) } +func TestWatcherObserverRunsBeforeDebounceAndSurvivesClose(t *testing.T) { + localDir := t.TempDir() + observed := make(chan string, 1) + callbacks := make(chan watcherEvent, 1) + watcher, err := NewFileWatcher(localDir, func(relativePath string, op fsnotify.Op) { + callbacks <- watcherEvent{path: relativePath, op: op} + }) + if err != nil { + t.Fatalf("create file watcher: %v", err) + } + watcher.onObserve = func(relativePath string) { + observed <- filepath.ToSlash(relativePath) + } + + watcher.queueChange("final-reply.json", fsnotify.Write) + select { + case got := <-observed: + if got != "final-reply.json" { + t.Fatalf("observed path = %q", got) + } + default: + t.Fatal("observer did not run synchronously before debounce") + } + if err := watcher.Close(); err != nil { + t.Fatalf("close watcher: %v", err) + } + assertNoWatcherEvents(t, callbacks, 150*time.Millisecond) +} + func TestWatcherStartReturnsLimitExceededWhenDirectoryBudgetExceeded(t *testing.T) { t.Setenv("RELAYFILE_MOUNT_MAX_WATCH_DIRS", "1") localDir := t.TempDir()