diff --git a/internal/mountsync/http_client_test.go b/internal/mountsync/http_client_test.go index e01562c1..063ac17a 100644 --- a/internal/mountsync/http_client_test.go +++ b/internal/mountsync/http_client_test.go @@ -14,6 +14,41 @@ import ( "testing" ) +func TestHTTPClientListTreeRequestsPrePaginationMountRuntimeExclusion(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/workspaces/ws_mount/fs/tree" { + t.Fatalf("unexpected path %q", r.URL.Path) + } + query := r.URL.Query() + if got := query.Get("path"); got != "/slack/channels" { + t.Fatalf("expected normalized mount path, got %q", got) + } + if got := query.Get("depth"); got != "3" { + t.Fatalf("expected depth 3, got %q", got) + } + if got := query.Get("cursor"); got != "/slack/channels/C123" { + t.Fatalf("expected cursor to be forwarded, got %q", got) + } + if got := query.Get("excludeMountRuntime"); got != "true" { + t.Fatalf("expected server-side mount runtime exclusion, got %q", got) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"path":"/slack/channels","entries":[],"nextCursor":null}`)) + })) + defer server.Close() + + client := NewHTTPClient(server.URL, "token", server.Client()) + if _, err := client.ListTree( + context.Background(), + "ws_mount", + "/slack/channels/", + 3, + "/slack/channels/C123", + ); err != nil { + t.Fatalf("list tree failed: %v", err) + } +} + func TestHTTPClientRetriesTransientFailure(t *testing.T) { var calls int32 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/internal/mountsync/syncer.go b/internal/mountsync/syncer.go index 6b1cb045..8165fb36 100644 --- a/internal/mountsync/syncer.go +++ b/internal/mountsync/syncer.go @@ -113,12 +113,18 @@ const ( defaultCursorResolutionAttempts = 3 defaultCursorRetryBaseDelay = 250 * time.Millisecond defaultBootstrapReadWorkers = 16 - defaultIncrementalEventPageLimit = 50 - defaultRetryAfterMaxDelay = 60 * time.Second - defaultWebSocketReconnectBase = 1 * time.Second - defaultWebSocketReconnectMax = 60 * time.Second - defaultWebSocketReconnectJitter = 200 * time.Millisecond - DefaultWebSocketMaintenanceEvery = 1 * time.Second + // fullTreeTraversalDepth bounds each tree request so the client can see and + // prune a reserved .relay directory before the server reaches deep + // outbox/acked/mountcmd_* leaves. Depth 3 also covers the common Slack + // channels//messages frontier in one request, avoiding one network + // round trip per message directory. + fullTreeTraversalDepth = 3 + defaultIncrementalEventPageLimit = 50 + defaultRetryAfterMaxDelay = 60 * time.Second + defaultWebSocketReconnectBase = 1 * time.Second + defaultWebSocketReconnectMax = 60 * time.Second + defaultWebSocketReconnectJitter = 200 * time.Millisecond + DefaultWebSocketMaintenanceEvery = 1 * time.Second ) // resolveDurationEnv returns the option value if non-zero, else parses the @@ -417,6 +423,10 @@ func (c *HTTPClient) logHTTPStatus(method, requestPath string, statusCode int, r func (c *HTTPClient) ListTree(ctx context.Context, workspaceID, path string, depth int, cursor string) (TreeResponse, error) { q := url.Values{} q.Set("path", normalizeRemotePath(path)) + // Relayfile mount owns .relay and its atomic state files as private runtime + // data. Ask the server to remove those rows before raw SQL pagination so a + // large acked outbox cannot consume every page before real content appears. + q.Set("excludeMountRuntime", "true") if depth > 0 { q.Set("depth", fmt.Sprintf("%d", depth)) } @@ -1127,11 +1137,12 @@ type mountState struct { // are additive/omitempty: legacy state files load with zero values // (BootstrapComplete=false), which self-heals by forcing a full // reconcile on the next cycle. - BootstrapComplete bool `json:"bootstrapComplete,omitempty"` - BootstrapCursor string `json:"bootstrapCursor,omitempty"` - BootstrapFilesSynced int `json:"bootstrapFilesSynced,omitempty"` - BootstrapFilesTotal int `json:"bootstrapFilesTotal,omitempty"` - BootstrapStartedAt string `json:"bootstrapStartedAt,omitempty"` + BootstrapComplete bool `json:"bootstrapComplete,omitempty"` + BootstrapDirectories []string `json:"bootstrapDirectories,omitempty"` + BootstrapCursor string `json:"bootstrapCursor,omitempty"` + BootstrapFilesSynced int `json:"bootstrapFilesSynced,omitempty"` + BootstrapFilesTotal int `json:"bootstrapFilesTotal,omitempty"` + BootstrapStartedAt string `json:"bootstrapStartedAt,omitempty"` // QuarantinedPaths holds remote paths that cannot be materialized locally // due to file/directory name collisions. Persisted across cycle restarts // so the daemon does not re-fetch these paths from the cloud until the @@ -4027,33 +4038,87 @@ func isLazyGithubRepoSubtreePath(path string) bool { return false } -func (s *Syncer) pullRemoteFullTree(ctx context.Context, conflicted map[string]struct{}, prog bootstrapProgress) error { +func (s *Syncer) pullRemoteFullTree(ctx context.Context, conflicted map[string]struct{}, prog bootstrapProgress) (returnErr error) { + metrics := fullTreeTraversalMetrics{startedAt: time.Now()} + defer func() { + s.logf( + "mount full-tree traversal summary remote_root=%q list_calls=%d entries_seen=%d files_seen=%d directories_seen=%d bytes_seen=%d runtime_entries_seen=%d runtime_subtrees_pruned=%d traversal_complete=%t traversal_failed=%t duration_ms=%d", + s.remoteRoot, + metrics.listCalls, + metrics.entriesSeen, + metrics.filesSeen, + metrics.directoriesSeen, + metrics.bytesSeen, + metrics.runtimeEntriesSeen, + metrics.runtimeSubtreesPruned, + metrics.traversalComplete, + returnErr != nil, + time.Since(metrics.startedAt).Milliseconds(), + ) + }() remotePaths := map[string]struct{}{} - // Resumable bootstrap: if a prior bootstrap was interrupted mid-tree, - // pick traversal back up from the persisted cursor rather than - // re-reading everything. startedFromEmpty tracks whether this process - // traversed the WHOLE tree (empty start -> NextCursor==nil): only then - // is the snapshot delete pass authoritative. Resuming from a persisted - // cursor means we did not observe the full remote set this cycle, so - // the delete pass is skipped (next full cycle does the authoritative - // delete) — preserving the #164/#165 mount-root-clobber invariants. + // Traverse in bounded-depth chunks. A depth=200 ListTree request makes the + // server enumerate every descendant before the client can reject reserved + // runtime paths. A small fixed depth sees a nested .relay directory before + // it reaches deep outbox/acked/mountcmd_* leaves, while advancing through a + // representative Slack tree in a few requests rather than one per directory. + directories := []string{s.remoteRoot} cursor := "" - if !s.state.BootstrapComplete && strings.TrimSpace(s.state.BootstrapCursor) != "" { - cursor = s.state.BootstrapCursor - s.logf("resuming bootstrap full-tree pull from persisted cursor (%d files already synced)", s.state.BootstrapFilesSynced) + startedFromEmpty := true + if !s.state.BootstrapComplete && len(s.state.BootstrapDirectories) > 0 { + resumedDirectories := make([]string, 0, len(s.state.BootstrapDirectories)) + for _, directory := range s.state.BootstrapDirectories { + directory = normalizeRemotePath(directory) + if !isUnderRemoteRoot(s.remoteRoot, directory) || isMountRuntimeRemotePath(directory) { + continue + } + resumedDirectories = append(resumedDirectories, directory) + } + if len(resumedDirectories) > 0 { + directories = resumedDirectories + cursor = strings.TrimSpace(s.state.BootstrapCursor) + startedFromEmpty = false + s.logf("resuming bootstrap bounded-tree pull at %s from persisted cursor (%d directories pending, %d files already synced)", directories[0], len(directories), s.state.BootstrapFilesSynced) + } + } else if !s.state.BootstrapComplete && strings.TrimSpace(s.state.BootstrapCursor) != "" { + // v0.10.28 and older persisted a cursor into one depth=200 listing. + // That cursor is invalid for a bounded-depth directory page, so restart from + // the root. The local-hash fast path avoids re-reading unchanged files. + s.logf("discarding legacy deep-tree bootstrap cursor and restarting bounded traversal from %s", s.remoteRoot) + s.state.BootstrapCursor = "" + } + queuedDirectories := make(map[string]struct{}, len(directories)) + for _, directory := range directories { + queuedDirectories[directory] = struct{}{} } - startedFromEmpty := cursor == "" if s.state.BootstrapStartedAt == "" { s.state.BootstrapStartedAt = time.Now().UTC().Format(time.RFC3339Nano) } + persistTraversal := func(filesThisPage int) error { + if s.state.BootstrapComplete { + return nil + } + s.state.BootstrapDirectories = append([]string(nil), directories...) + s.state.BootstrapCursor = cursor + s.state.BootstrapFilesSynced += filesThisPage + prog.touch() + if err := s.saveState(); err != nil { + return err + } + prog.touch() + return nil + } maxObservedRevision := "" var transientBootstrapAbort bool - for { + prunedRuntimeRoots := map[string]struct{}{} + for len(directories) > 0 { + currentDirectory := directories[0] var page TreeResponse var err error s.runFullPullIO(func() { - page, err = s.client.ListTree(ctx, s.workspace, s.remoteRoot, 200, cursor) + page, err = s.client.ListTree(ctx, s.workspace, currentDirectory, fullTreeTraversalDepth, cursor) }) + metrics.listCalls++ if err != nil { s.recordCloudFailure(err) return err @@ -4063,23 +4128,52 @@ func (s *Syncer) pullRemoteFullTree(ctx context.Context, conflicted map[string]s filesThisPage := 0 readJobs := make([]bootstrapReadJob, 0, len(page.Entries)) for _, entry := range page.Entries { - if entry.Type != "file" { - continue + remotePath := normalizeRemotePath(entry.Path) + metrics.entriesSeen++ + switch entry.Type { + case "file": + metrics.filesSeen++ + if entry.Size > 0 { + metrics.bytesSeen += entry.Size + } + case "dir": + metrics.directoriesSeen++ } - if revisionAdvances(maxObservedRevision, entry.Revision) { - maxObservedRevision = entry.Revision + runtimeRoot := mountRuntimeRemoteRoot(remotePath) + if runtimeRoot != "" { + metrics.runtimeEntriesSeen++ } - remotePath := normalizeRemotePath(entry.Path) if !isUnderRemoteRoot(s.remoteRoot, remotePath) { continue } - if s.skipMountRuntimeRemotePath(remotePath, conflicted) { + if runtimeRoot != "" { + if _, pruned := prunedRuntimeRoots[runtimeRoot]; !pruned { + prunedRuntimeRoots[runtimeRoot] = struct{}{} + metrics.runtimeSubtreesPruned++ + s.skipMountRuntimeRemotePath(runtimeRoot, conflicted) + } continue } // Contract: lazy GitHub repos do not eagerly hydrate per-repo content at startup. if s.lazyRepos && isUnderLazyGithubRepoSubtree(s.remoteRoot, remotePath) { continue } + if entry.Type == "dir" { + if remoteDescendantDepth(currentDirectory, remotePath) >= fullTreeTraversalDepth { + if _, exists := queuedDirectories[remotePath]; exists { + continue + } + queuedDirectories[remotePath] = struct{}{} + directories = append(directories, remotePath) + } + continue + } + if entry.Type != "file" { + continue + } + if revisionAdvances(maxObservedRevision, entry.Revision) { + maxObservedRevision = entry.Revision + } if tracked, ok := s.state.Files[remotePath]; ok && tracked.Denied { continue } @@ -4174,30 +4268,30 @@ func (s *Syncer) pullRemoteFullTree(ctx context.Context, conflicted map[string]s // cursor so the same page (including the failing path) is retried next // cycle. Save state up to the last successfully-committed cursor. if transientBootstrapAbort { - if !s.state.BootstrapComplete { - s.state.BootstrapFilesSynced += filesThisPage - if err := s.saveState(); err != nil { - return err - } + if err := persistTraversal(filesThisPage); err != nil { + return err } break } - if page.NextCursor == nil || *page.NextCursor == "" { - break + if page.NextCursor != nil && *page.NextCursor != "" { + cursor = *page.NextCursor + if err := persistTraversal(filesThisPage); err != nil { + return err + } + continue } - cursor = *page.NextCursor - // Persist the resume point + progress so an interrupted bootstrap - // (timeout, crash, watchdog cancel) picks up here next cycle - // instead of restarting the whole tree. - if !s.state.BootstrapComplete { - s.state.BootstrapCursor = cursor - s.state.BootstrapFilesSynced += filesThisPage - prog.touch() - if err := s.saveState(); err != nil { + // This directory is exhausted. Persist the remaining queue before + // requesting its next member so cancellation resumes at a directory + // boundary instead of restarting the root. + directories = directories[1:] + cursor = "" + if len(directories) > 0 { + if err := persistTraversal(filesThisPage); err != nil { return err } - prog.touch() + continue } + metrics.traversalComplete = true } // A transient read error stopped the page scan early. Bootstrap is not @@ -4210,8 +4304,8 @@ func (s *Syncer) pullRemoteFullTree(ctx context.Context, conflicted map[string]s // Resumed/partial traversal safety: only run the authoritative // snapshot delete pass when this process traversed the ENTIRE tree - // (started from an empty cursor and reached NextCursor==nil). If the - // traversal began from a persisted resume cursor, remotePaths only + // (started from the root with an empty directory queue). If the + // traversal began from a persisted directory/cursor boundary, remotePaths only // covers the tail of the tree, so deleting "everything not in // remotePaths" would wipe the already-mirrored prefix — exactly the // #164/#165 clobber failure mode. Skip the delete pass this cycle; @@ -4244,6 +4338,18 @@ func (s *Syncer) pullRemoteFullTree(ctx context.Context, conflicted map[string]s return nil } +type fullTreeTraversalMetrics struct { + startedAt time.Time + listCalls int + entriesSeen int + filesSeen int + directoriesSeen int + bytesSeen int64 + runtimeEntriesSeen int + runtimeSubtreesPruned int + traversalComplete bool +} + type bootstrapReadJob struct { Index int RemotePath string @@ -4368,6 +4474,7 @@ func bootstrapReadWorkers() int { // gate (Step 5) cannot be satisfied by a partial pull. func (s *Syncer) markBootstrapComplete() { s.state.BootstrapComplete = true + s.state.BootstrapDirectories = nil s.state.BootstrapCursor = "" s.state.BootstrapStartedAt = "" s.state.BootstrapFilesSynced = 0 @@ -5450,24 +5557,58 @@ func (s *Syncer) applyRemoteFile(remotePath string, file RemoteFile, conflicted } func (s *Syncer) skipMountRuntimeRemotePath(remotePath string, conflicted map[string]struct{}) bool { - remotePath = normalizeRemotePath(remotePath) - if !isMountRuntimeRemotePath(remotePath) { + remotePath = mountRuntimeRemoteRoot(remotePath) + if remotePath == "" { return false } s.logf("skipping mount runtime path surfaced as workspace content: %s", remotePath) - if err := s.applyRemoteDelete(remotePath, conflicted); err != nil { - s.logf("failed to clean local mount runtime path %s: %v", remotePath, err) + prefix := strings.TrimSuffix(remotePath, "/") + "/" + for trackedPath := range s.state.Files { + if trackedPath != remotePath && !strings.HasPrefix(trackedPath, prefix) { + continue + } + delete(s.state.Files, trackedPath) + s.clearIncrementalReadNotReady(trackedPath) + if conflicted != nil { + delete(conflicted, trackedPath) + } + } + for notReadyPath := range s.state.IncrementalReadNotReadySince { + if notReadyPath == remotePath || strings.HasPrefix(notReadyPath, prefix) { + s.clearIncrementalReadNotReady(notReadyPath) + } } if localPath, err := s.remoteToLocalPath(remotePath); err == nil { - if removeErr := os.Remove(localPath); removeErr != nil && !errors.Is(removeErr, os.ErrNotExist) { - s.logf("failed to remove local mount runtime file %s (%s): %v", remotePath, localPath, removeErr) + if s.isActiveMountRuntimeLocalPath(localPath) { + // A leaked remote /.relay entry maps onto this daemon's own public + // runtime directory. Reject the remote entry, but never delete the + // live local state/outbox while doing so. + } else if err := s.assertNotMountRoot(localPath); err != nil { + s.logf("failed to clean local mount runtime path %s: %v", remotePath, err) + } else if removeErr := os.RemoveAll(localPath); removeErr != nil { + s.logf("failed to remove local mount runtime subtree %s (%s): %v", remotePath, localPath, removeErr) } } - delete(s.state.Files, remotePath) s.clearIncrementalReadNotReady(remotePath) return true } +func (s *Syncer) isActiveMountRuntimeLocalPath(localPath string) bool { + localPath = filepath.Clean(localPath) + publicRuntimeRoot := filepath.Join(filepath.Clean(s.localRoot), ".relay") + if relative, err := filepath.Rel(publicRuntimeRoot, localPath); err == nil && + (relative == "." || (relative != ".." && !strings.HasPrefix(relative, ".."+string(filepath.Separator)))) { + return true + } + stateFile := filepath.Clean(s.stateFile) + if localPath == stateFile { + return true + } + stateTempPrefix := strings.TrimSuffix(atomicTempPattern(stateFile), "*") + return filepath.Dir(localPath) == filepath.Dir(stateFile) && + strings.HasPrefix(filepath.Base(localPath), stateTempPrefix) +} + func (s *Syncer) applyLocalPermissions(localPath string, canWrite bool) error { if canWrite { return os.Chmod(localPath, 0o644) @@ -6101,25 +6242,34 @@ func normalizeScopes(scopes []string) []string { } func isMountRuntimeRemotePath(path string) bool { - return isMountRuntimeRelativePath(strings.TrimPrefix(normalizeRemotePath(path), "/")) + return mountRuntimeRemoteRoot(path) != "" } func isMountRuntimeRelativePath(path string) bool { - normalized := filepath.ToSlash(strings.TrimSpace(path)) + return mountRuntimeRemoteRoot("/"+strings.TrimPrefix(filepath.ToSlash(strings.TrimSpace(path)), "/")) != "" +} + +// mountRuntimeRemoteRoot returns the reserved runtime subtree containing path. +// Canonicalizing descendants to their first reserved segment lets a bounded +// tree page clean/log one .relay root without treating its state/outbox +// children as separate subtrees. +func mountRuntimeRemoteRoot(path string) string { + normalized := strings.TrimPrefix(filepath.ToSlash(normalizeRemotePath(path)), "/") if normalized == "" || normalized == "." { - return false + return "" } - for _, segment := range strings.Split(normalized, "/") { + segments := strings.Split(normalized, "/") + for index, segment := range segments { if segment == "" || segment == "." { continue } if segment == ".relay" || segment == ".relayfile-mount-state.json" || strings.HasPrefix(segment, ".relayfile-mount-state.json.tmp-") { - return true + return normalizeRemotePath("/" + strings.Join(segments[:index+1], "/")) } } - return false + return "" } func (s *Syncer) logMountControlPathSkipped(relativePath string) { @@ -6161,6 +6311,7 @@ func (s *Syncer) loadState() error { if s.state.BootstrapComplete && s.state.SyncMode == "write-only" && !s.writeOnly { s.logf("syncMode transition write-only->mirror detected; resetting BootstrapComplete to force a full bootstrap pull (backfills records missed while write-only)") s.state.BootstrapComplete = false + s.state.BootstrapDirectories = nil s.state.BootstrapCursor = "" s.state.BootstrapStartedAt = "" s.state.BootstrapFilesSynced = 0 @@ -6728,6 +6879,20 @@ func isUnderRemoteRoot(remoteRoot, remotePath string) bool { return remotePath == remoteRoot || strings.HasPrefix(remotePath, remoteRoot+"/") } +func remoteDescendantDepth(remoteRoot, remotePath string) int { + remoteRoot = normalizeRemotePath(remoteRoot) + remotePath = normalizeRemotePath(remotePath) + if remotePath == remoteRoot || !isUnderRemoteRoot(remoteRoot, remotePath) { + return 0 + } + relative := strings.TrimPrefix(remotePath, remoteRoot) + relative = strings.TrimPrefix(relative, "/") + if relative == "" { + return 0 + } + return len(strings.Split(relative, "/")) +} + func remoteToLocalPath(localRoot, remoteRoot, remotePath string) (string, error) { localRoot = filepath.Clean(localRoot) remoteRoot = normalizeRemotePath(remoteRoot) diff --git a/internal/mountsync/syncer_test.go b/internal/mountsync/syncer_test.go index d6c77dbf..d531da67 100644 --- a/internal/mountsync/syncer_test.go +++ b/internal/mountsync/syncer_test.go @@ -7477,6 +7477,7 @@ func (c *fakeClient) ListTree(ctx context.Context, workspaceID, path string, dep Type: "file", Revision: file.Revision, ContentHash: file.ContentHash, + Size: int64(len(file.Content)), }) } sort.Slice(entries, func(i, j int) bool { return entries[i].Path < entries[j].Path }) @@ -7487,6 +7488,99 @@ func (c *fakeClient) ListTree(ctx context.Context, workspaceID, path string, dep }, nil } +type hierarchicalTreeCall struct { + path string + depth int + entries int +} + +// hierarchicalTreeClient mirrors the production tree contract: a request +// returns files and synthesized directories only through the requested depth. +// It lets traversal tests distinguish pruning before descent from filtering a +// flat depth=200 response after the server has already enumerated every leaf. +type hierarchicalTreeClient struct { + *fakeClient + calls []hierarchicalTreeCall + entriesReturned int + returnedPaths []string + listDelay time.Duration + failAtCall int + failErr error +} + +func (c *hierarchicalTreeClient) ListTree(ctx context.Context, _ string, path string, depth int, cursor string) (TreeResponse, error) { + base := normalizeRemotePath(path) + if depth <= 0 { + depth = 1 + } + if c.listDelay > 0 { + select { + case <-time.After(c.listDelay): + case <-ctx.Done(): + return TreeResponse{}, ctx.Err() + } + } + callNumber := len(c.calls) + 1 + if c.failAtCall == callNumber { + c.calls = append(c.calls, hierarchicalTreeCall{path: base, depth: depth}) + c.failAtCall = 0 + if c.failErr != nil { + return TreeResponse{}, c.failErr + } + return TreeResponse{}, errors.New("synthetic ListTree failure") + } + entriesByPath := map[string]TreeEntry{} + for remotePath, file := range c.files { + remotePath = normalizeRemotePath(remotePath) + if !isUnderRemoteRoot(base, remotePath) || remotePath == base { + continue + } + relative := strings.TrimPrefix(strings.TrimPrefix(remotePath, base), "/") + parts := strings.Split(relative, "/") + levels := depth + if len(parts) < levels { + levels = len(parts) + } + for level := 1; level <= levels; level++ { + entryPath := normalizeRemotePath(strings.TrimSuffix(base, "/") + "/" + strings.Join(parts[:level], "/")) + if level == len(parts) { + entriesByPath[entryPath] = TreeEntry{ + Path: entryPath, + Type: "file", + Revision: file.Revision, + ContentHash: file.ContentHash, + Size: int64(len(file.Content)), + } + continue + } + if _, exists := entriesByPath[entryPath]; !exists { + entriesByPath[entryPath] = TreeEntry{Path: entryPath, Type: "dir", Revision: "dir"} + } + } + } + entries := make([]TreeEntry, 0, len(entriesByPath)) + for _, entry := range entriesByPath { + entries = append(entries, entry) + } + sort.Slice(entries, func(i, j int) bool { return entries[i].Path < entries[j].Path }) + if cursor != "" { + start := len(entries) + for index, entry := range entries { + if entry.Path == cursor { + start = index + 1 + break + } + } + entries = entries[start:] + } + c.calls = append(c.calls, hierarchicalTreeCall{path: base, depth: depth, entries: len(entries)}) + c.entriesReturned += len(entries) + for _, entry := range entries { + c.returnedPaths = append(c.returnedPaths, entry.Path) + } + return TreeResponse{Path: base, Entries: entries}, nil +} + func (c *fakeClient) LatestEventID(ctx context.Context, workspaceID, provider string) (string, error) { _ = ctx _ = workspaceID @@ -9749,6 +9843,9 @@ func TestLoadStateResetsBootstrapCompleteOnlyOnWriteOnlyToMirrorSyncMode(t *test t.Fatalf("BootstrapComplete = %v, want %v", got, tc.wantBootstrapComplete) } if tc.wantCleared { + if len(syncer.state.BootstrapDirectories) != 0 { + t.Fatalf("BootstrapDirectories = %#v, want empty", syncer.state.BootstrapDirectories) + } if syncer.state.BootstrapCursor != "" { t.Fatalf("BootstrapCursor = %q, want empty", syncer.state.BootstrapCursor) } @@ -9880,10 +9977,12 @@ func TestPullRemoteFullTreeSkipsNestedMountRuntimeState(t *testing.T) { }, } localDir := t.TempDir() + logger := &captureLogger{} syncer, err := NewSyncer(client, SyncerOptions{ WorkspaceID: "ws_pull_nested_relay", RemoteRoot: "/slack", LocalRoot: localDir, + Logger: logger, }) if err != nil { t.Fatalf("new syncer failed: %v", err) @@ -9917,6 +10016,269 @@ func TestPullRemoteFullTreeSkipsNestedMountRuntimeState(t *testing.T) { if got := client.readFileCallsByPath["/slack/channels/C123/messages/.relay/state.json"]; got != 0 { t.Fatalf("nested mount runtime path should not be read from remote, got %d reads", got) } + var summary string + for _, line := range logger.lines { + if strings.Contains(line, "mount full-tree traversal summary") { + summary = line + } + } + if summary == "" { + t.Fatalf("expected a full-tree traversal summary, logs: %#v", logger.lines) + } + for _, field := range []string{ + "list_calls=1", + "entries_seen=2", + "files_seen=2", + "directories_seen=0", + "runtime_entries_seen=1", + "traversal_complete=true", + } { + if !strings.Contains(summary, field) { + t.Fatalf("traversal summary %q missing %q", summary, field) + } + } +} + +func TestIsMountRuntimeRelativePathExcludesOnlyReservedRuntimeArtifacts(t *testing.T) { + tests := []struct { + path string + want bool + }{ + {path: ".relay", want: true}, + {path: "slack/channels/C123/messages/1/.relay/state.json", want: true}, + {path: "slack/channels/C123/messages/1/.relay/outbox/acked/mountcmd_1.json", want: true}, + {path: "slack/channels/C123/messages/1/.relayfile-mount-state.json", want: true}, + {path: "slack/channels/C123/messages/1/relay/state.json", want: false}, + {path: "slack/channels/C123/messages/1/.relay-notes/state.json", want: false}, + {path: "slack/channels/C123/messages/1/outbox/acked/mountcmd_1.json", want: false}, + {path: "slack/channels/C123/messages/1/notes/mountcmd_1.json", want: false}, + {path: "slack/channels/C123/messages/1/.relayfile-mount-state.json.backup", want: false}, + } + for _, tt := range tests { + t.Run(tt.path, func(t *testing.T) { + if got := isMountRuntimeRelativePath(tt.path); got != tt.want { + t.Fatalf("isMountRuntimeRelativePath(%q) = %t, want %t", tt.path, got, tt.want) + } + }) + } +} + +func TestSkipMountRuntimeRemotePathPreservesActiveRootRuntime(t *testing.T) { + localDir := t.TempDir() + activeState := filepath.Join(localDir, ".relay", "state.json") + activeOutbox := filepath.Join(localDir, ".relay", "outbox", "pending", "mountcmd_live.json") + activeLegacyState := filepath.Join(localDir, LegacyMountStateFileName) + activeLegacyTemp := filepath.Join(localDir, ".relayfile-mount-state.json.tmp-live") + if err := os.MkdirAll(filepath.Dir(activeOutbox), 0o755); err != nil { + t.Fatalf("mkdir active runtime tree: %v", err) + } + if err := os.WriteFile(activeState, []byte(`{"ready":true}`), 0o644); err != nil { + t.Fatalf("write active state: %v", err) + } + if err := os.WriteFile(activeOutbox, []byte(`{"pending":true}`), 0o644); err != nil { + t.Fatalf("write active outbox: %v", err) + } + if err := os.WriteFile(activeLegacyState, []byte(`{"ready":true}`), 0o644); err != nil { + t.Fatalf("write active legacy state: %v", err) + } + if err := os.WriteFile(activeLegacyTemp, []byte(`{"pending":true}`), 0o644); err != nil { + t.Fatalf("write active legacy state temp: %v", err) + } + syncer, err := NewSyncer(&fakeClient{files: map[string]RemoteFile{}}, SyncerOptions{ + WorkspaceID: "ws_active_runtime", + RemoteRoot: "/", + LocalRoot: localDir, + }) + if err != nil { + t.Fatalf("new syncer failed: %v", err) + } + for _, remoteRuntimePath := range []string{ + "/.relay", + "/.relayfile-mount-state.json", + "/.relayfile-mount-state.json.tmp-live", + } { + if !syncer.skipMountRuntimeRemotePath(remoteRuntimePath, nil) { + t.Fatalf("expected remote %s to be classified as runtime", remoteRuntimePath) + } + } + for _, activePath := range []string{activeState, activeOutbox, activeLegacyState, activeLegacyTemp} { + if _, err := os.Stat(activePath); err != nil { + t.Fatalf("active mount runtime %s was removed: %v", activePath, err) + } + } +} + +func TestPullRemoteFullTreePrunesNestedMountRuntimeBeforeDescendantEnumeration(t *testing.T) { + files := map[string]RemoteFile{ + "/slack/channels/C123/messages/1780145510_376649.json": { + Path: "/slack/channels/C123/messages/1780145510_376649.json", Revision: "rev_msg", Content: `{"text":"hello"}`, + }, + "/slack/channels/C123/messages/1780145510_376649/.relay-notes/keep.md": { + Path: "/slack/channels/C123/messages/1780145510_376649/.relay-notes/keep.md", Revision: "rev_notes", Content: "real notes", + }, + "/slack/channels/C123/messages/1780145510_376649/outbox/acked/mountcmd_real.json": { + Path: "/slack/channels/C123/messages/1780145510_376649/outbox/acked/mountcmd_real.json", Revision: "rev_real_cmd", Content: `{"real":true}`, + }, + "/slack/channels/C123/messages/1780145510_376649/.relayfile-mount-state.json.backup": { + Path: "/slack/channels/C123/messages/1780145510_376649/.relayfile-mount-state.json.backup", Revision: "rev_real_backup", Content: `{"backup":true}`, + }, + } + for message := 0; message < 200; message++ { + messageID := fmt.Sprintf("1780145510_%06d", message) + statePath := fmt.Sprintf("/slack/channels/C123/messages/%s/.relay/state.json", messageID) + files[statePath] = RemoteFile{Path: statePath, Revision: fmt.Sprintf("rev_runtime_state_%03d", message), Content: `{"pendingWriteback":2}`} + for artifact := 0; artifact < 2; artifact++ { + path := fmt.Sprintf("/slack/channels/C123/messages/%s/.relay/outbox/acked/mountcmd_%03d.json", messageID, artifact) + files[path] = RemoteFile{Path: path, Revision: fmt.Sprintf("rev_runtime_%03d_%d", message, artifact), Content: `{"runtime":true}`} + } + } + baseClient := &fakeClient{files: files} + client := &hierarchicalTreeClient{fakeClient: baseClient, listDelay: 5 * time.Millisecond} + logger := &captureLogger{} + localDir := t.TempDir() + syncer, err := NewSyncer(client, SyncerOptions{ + WorkspaceID: "ws_prune_nested_relay", + RemoteRoot: "/slack", + LocalRoot: localDir, + Logger: logger, + }) + if err != nil { + t.Fatalf("new syncer failed: %v", err) + } + runtimeDir := filepath.Join(localDir, "channels", "C123", "messages", "1780145510_000000", ".relay") + staleRuntimePath := filepath.Join(runtimeDir, "outbox", "acked", "mountcmd_stale.json") + if err := os.MkdirAll(filepath.Dir(staleRuntimePath), 0o755); err != nil { + t.Fatalf("mkdir stale runtime tree: %v", err) + } + if err := os.WriteFile(staleRuntimePath, []byte(`{"stale":true}`), 0o644); err != nil { + t.Fatalf("write stale runtime artifact: %v", err) + } + staleRemotePath := "/slack/channels/C123/messages/1780145510_000000/.relay/outbox/acked/mountcmd_stale.json" + syncer.state.Files[staleRemotePath] = trackedFile{Revision: "rev_stale", Hash: hashString(`{"stale":true}`)} + + startedAt := time.Now() + if err := syncer.pullRemoteFullTree(context.Background(), nil, bootstrapProgress{}); err != nil { + t.Fatalf("pullRemoteFullTree failed: %v", err) + } + elapsed := time.Since(startedAt) + + if len(client.calls) != 3 { + t.Fatalf("representative Slack traversal used %d ListTree calls, want exactly 3: %#v", len(client.calls), client.calls) + } + if elapsed >= 500*time.Millisecond { + t.Fatalf("representative Slack traversal exceeded bounded latency: %s across %d delayed calls", elapsed, len(client.calls)) + } + if client.entriesReturned >= 1000 { + t.Fatalf("runtime subtree traversal was not bounded: returned %d entries across calls %#v", client.entriesReturned, client.calls) + } + for _, call := range client.calls { + if isMountRuntimeRemotePath(call.path) { + t.Fatalf("ListTree descended into reserved runtime subtree: %#v", call) + } + if call.depth != fullTreeTraversalDepth { + t.Fatalf("ListTree depth = %d for %s, want bounded depth %d", call.depth, call.path, fullTreeTraversalDepth) + } + } + for _, returnedPath := range client.returnedPaths { + if strings.Contains(filepath.Base(returnedPath), "mountcmd_") && isMountRuntimeRemotePath(returnedPath) { + t.Fatalf("deep runtime artifact was enumerated before pruning: %s", returnedPath) + } + } + for _, realPath := range []string{ + filepath.Join(localDir, "channels", "C123", "messages", "1780145510_376649.json"), + filepath.Join(localDir, "channels", "C123", "messages", "1780145510_376649", ".relay-notes", "keep.md"), + filepath.Join(localDir, "channels", "C123", "messages", "1780145510_376649", "outbox", "acked", "mountcmd_real.json"), + filepath.Join(localDir, "channels", "C123", "messages", "1780145510_376649", ".relayfile-mount-state.json.backup"), + } { + if _, err := os.Stat(realPath); err != nil { + t.Fatalf("expected real workspace content %s to remain mirrored: %v", realPath, err) + } + } + assertLocalFileContent(t, filepath.Join(localDir, "channels", "C123", "messages", "1780145510_376649", ".relayfile-mount-state.json.backup"), `{"backup":true}`) + if _, err := os.Stat(runtimeDir); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("nested runtime subtree should be absent after pruning, stat err=%v", err) + } + if _, exists := syncer.state.Files[staleRemotePath]; exists { + t.Fatalf("stale nested runtime artifact remained tracked") + } + for path, calls := range baseClient.readFileCallsByPath { + if isMountRuntimeRemotePath(path) && calls > 0 { + t.Fatalf("runtime artifact %s was read %d time(s)", path, calls) + } + } + var summary string + for _, line := range logger.lines { + if strings.Contains(line, "mount full-tree traversal summary") { + summary = line + } + } + if !strings.Contains(summary, "runtime_subtrees_pruned=200") { + t.Fatalf("expected traversal summary to report 200 pruned runtime subtrees, got %q", summary) + } + if !strings.Contains(summary, "list_calls=3") { + t.Fatalf("expected traversal summary to report the measured three-call frontier, got %q", summary) + } +} + +func TestPullRemoteFullTreeResumesPersistedShallowDirectoryQueue(t *testing.T) { + files := map[string]RemoteFile{ + "/a/level1/level2/one.md": {Path: "/a/level1/level2/one.md", Revision: "rev_a", Content: "one"}, + "/b/level1/level2/two.md": {Path: "/b/level1/level2/two.md", Revision: "rev_b", Content: "two"}, + "/c/level1/level2/three.md": {Path: "/c/level1/level2/three.md", Revision: "rev_c", Content: "three"}, + } + client := &hierarchicalTreeClient{ + fakeClient: &fakeClient{files: files}, + failAtCall: 3, + failErr: &HTTPError{StatusCode: http.StatusBadGateway, Code: "bad_gateway", Message: "synthetic interruption"}, + } + localDir := t.TempDir() + syncer, err := NewSyncer(client, SyncerOptions{ + WorkspaceID: "ws_shallow_resume", + RemoteRoot: "/", + LocalRoot: localDir, + }) + if err != nil { + t.Fatalf("new syncer failed: %v", err) + } + preservedPrefixPath := filepath.Join(localDir, "preserved-prefix.md") + if err := os.WriteFile(preservedPrefixPath, []byte("prefix mirrored before interruption"), 0o644); err != nil { + t.Fatalf("seed tracked prefix: %v", err) + } + syncer.state.Files["/preserved-prefix.md"] = trackedFile{ + Revision: "rev_prefix", + Hash: hashString("prefix mirrored before interruption"), + } + if err := syncer.pullRemoteFullTree(context.Background(), nil, bootstrapProgress{}); err == nil { + t.Fatalf("expected first traversal to fail") + } + if got := syncer.state.BootstrapDirectories; len(got) == 0 || got[0] != "/b/level1/level2" { + t.Fatalf("persisted directory queue = %#v, want /b/level1/level2 at the resume boundary", got) + } + if err := syncer.pullRemoteFullTree(context.Background(), nil, bootstrapProgress{}); err != nil { + t.Fatalf("resumed traversal failed: %v", err) + } + if len(client.calls) < 4 || client.calls[3].path != "/b/level1/level2" { + t.Fatalf("resumed traversal did not restart at /b/level1/level2: %#v", client.calls) + } + if !syncer.state.BootstrapComplete { + t.Fatalf("resumed traversal did not mark bootstrap complete") + } + if len(syncer.state.BootstrapDirectories) != 0 { + t.Fatalf("completed traversal retained directory queue: %#v", syncer.state.BootstrapDirectories) + } + if _, tracked := syncer.state.Files["/preserved-prefix.md"]; !tracked { + t.Fatalf("resumed partial traversal applied an authoritative delete to the tracked prefix") + } + assertLocalFileContent(t, preservedPrefixPath, "prefix mirrored before interruption") + for _, localPath := range []string{ + filepath.Join(localDir, "a", "level1", "level2", "one.md"), + filepath.Join(localDir, "b", "level1", "level2", "two.md"), + filepath.Join(localDir, "c", "level1", "level2", "three.md"), + } { + if _, err := os.Stat(localPath); err != nil { + t.Fatalf("expected resumed traversal to mirror %s: %v", localPath, err) + } + } } func TestPullRemoteIncrementalSkipsNestedMountRuntimeState(t *testing.T) {