From f31608e786d89ee0ab5d0e7b30f16700dad0569d Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sat, 10 Oct 2026 15:53:27 -0700 Subject: [PATCH 1/2] perf(mountsync): linear bulk batch sizing, single read of new files, bounded concurrent uploads A 13k-file --push-local-once seed spent ~100s producing no requests and then sent its 8 MiB /fs/bulk chunks strictly one at a time. - chunkOutboxRecords, WriteFilesBulk and chunkPendingBulkWrites re-marshalled the whole pending batch for every appended file (O(n^2) bytes). Encode each file once and track the request size incrementally; boundaries are byte-identical to the old algorithm (kept as a test reference). - pushLocal's scan now reads small untracked files whole once (1 MiB per file, 64 MiB per scan, never in low-memory mode) instead of hashing and then re-reading them. Outbox durability is unchanged. - PushLocalAndFlushOnce derives the public state from pushLocal's scan rather than re-hashing the tree a third time. - Outbox flushes that drain several chunks pipeline up to 2 POSTs (RELAYFILE_MOUNT_BULK_UPLOAD_CONCURRENCY, 1 = serial). State transitions stay serial and in chunk order under s.mu; same-path chunks never overlap; acks only follow server responses; in-flight uploads are still applied after an error. Co-Authored-By: Claude Opus 5.5 --- docs/environment-variables.md | 1 + internal/mountsync/bulk_batch_sizing_test.go | 235 +++++++++ .../mountsync/bulk_upload_concurrency_test.go | 340 +++++++++++++ .../mountsync/push_local_read_once_test.go | 144 ++++++ internal/mountsync/push_perf_test.go | 216 +++++++++ internal/mountsync/stream_write.go | 45 +- internal/mountsync/syncer.go | 446 +++++++++++++++--- internal/mountsync/syncer_test.go | 6 +- 8 files changed, 1358 insertions(+), 75 deletions(-) create mode 100644 internal/mountsync/bulk_batch_sizing_test.go create mode 100644 internal/mountsync/bulk_upload_concurrency_test.go create mode 100644 internal/mountsync/push_local_read_once_test.go create mode 100644 internal/mountsync/push_perf_test.go diff --git a/docs/environment-variables.md b/docs/environment-variables.md index 3c2445c2..fd2cda27 100644 --- a/docs/environment-variables.md +++ b/docs/environment-variables.md @@ -147,6 +147,7 @@ Queue backend precedence is: | `RELAYFILE_MOUNT_INTERVAL_JITTER` | float | `0.2` | No | Clamped into the `0..1` range | | `RELAYFILE_MOUNT_TIMEOUT` | duration | `15s` | No | Per-sync timeout | | `RELAYFILE_MOUNT_WEBSOCKET` | bool | `true` | No | Enables WebSocket streaming when available | +| `RELAYFILE_MOUNT_BULK_UPLOAD_CONCURRENCY` | int | `2` | No | Maximum concurrent `/fs/bulk` POSTs while one outbox flush drains several request chunks; `1` restores strictly serial uploads; clamped to `8` | | `RELAYFILE_INCREMENTAL_READ_CONCURRENCY` | int | `16` | No | Bounded parallel reads for cursor/poll recovery; clamped to 64 | | `RELAYFILE_BOOTSTRAP_STALL_CYCLES` | int | `20` | No | Consecutive checkpoint-stable bootstrap cycles before the mount fails terminally | | `RELAYFILE_BOOTSTRAP_MAX_DIRECTORIES` | int | `50000` | No | Maximum distinct bounded-tree directories; raise only after inspecting the stalled path for cyclic/expanding aliases | diff --git a/internal/mountsync/bulk_batch_sizing_test.go b/internal/mountsync/bulk_batch_sizing_test.go new file mode 100644 index 00000000..0e120656 --- /dev/null +++ b/internal/mountsync/bulk_batch_sizing_test.go @@ -0,0 +1,235 @@ +package mountsync + +import ( + "encoding/base64" + "encoding/json" + "fmt" + "math/rand" + "strings" + "testing" +) + +// legacyBulkWriteRequestSize is the pre-incremental reference: marshal the +// whole request body and measure it. +func legacyBulkWriteRequestSize(files []BulkWriteFile) int64 { + body := struct { + Files []BulkWriteFile `json:"files"` + }{Files: files} + data, err := json.Marshal(body) + if err != nil { + return 0 + } + return int64(len(data)) +} + +// legacyChunkOutboxRecords is the O(n^2) reference implementation that +// chunkOutboxRecords replaced. Kept verbatim so the boundary test proves the +// incremental sizer is byte-identical. +func legacyChunkOutboxRecords(records []outboxRecord, maxBytes int64) [][]outboxRecord { + if len(records) == 0 { + return nil + } + if maxBytes <= 0 { + return [][]outboxRecord{records} + } + chunks := make([][]outboxRecord, 0, 1) + current := make([]outboxRecord, 0, len(records)) + for _, record := range records { + candidate := append(append([]outboxRecord(nil), current...), record) + if len(current) > 0 && legacyBulkWriteRequestSize(outboxRecordsAsBulkFiles(candidate)) > maxBytes { + chunks = append(chunks, append([]outboxRecord(nil), current...)) + current = current[:0] + } + current = append(current, record) + } + if len(current) > 0 { + chunks = append(chunks, current) + } + return chunks +} + +// legacyWriteFilesBulkBounds reproduces the pre-incremental WriteFilesBulk +// chunk loop and returns its [start,end) boundaries plus the request size the +// singleton streaming decision used. +func legacyWriteFilesBulkBounds(files []BulkWriteFile, maxBytes int64) []bulkWriteChunkBound { + var bounds []bulkWriteChunkBound + for start := 0; start < len(files); { + end := start + 1 + for end < len(files) && legacyBulkWriteRequestSize(files[start:end+1]) <= maxBytes { + end++ + } + bounds = append(bounds, bulkWriteChunkBound{start: start, end: end, size: legacyBulkWriteRequestSize(files[start:end])}) + start = end + } + return bounds +} + +func legacyChunkPendingBulkWrites(workspaceID string, pending []pendingBulkWrite, maxBytes int64) [][]pendingBulkWrite { + if len(pending) == 0 { + return nil + } + if maxBytes <= 0 { + return [][]pendingBulkWrite{pending} + } + chunks := make([][]pendingBulkWrite, 0, 1) + current := make([]pendingBulkWrite, 0, len(pending)) + for _, item := range pending { + candidate := append(append([]pendingBulkWrite(nil), current...), item) + if len(current) > 0 && legacyBulkWriteRequestSize(bulkWriteFilesForPending(workspaceID, candidate)) > maxBytes { + chunks = append(chunks, append([]pendingBulkWrite(nil), current...)) + current = current[:0] + } + current = append(current, item) + } + if len(current) > 0 { + chunks = append(chunks, current) + } + return chunks +} + +// mixedSizeOutboxCorpus builds records whose JSON encodings vary widely: +// empty, tiny, ~3 KiB code, HTML-escaped characters, U+2028, multibyte +// UTF-8, base64 binaries, symlinks, draft paths with content identities, and +// payloads both just under and well over the 8 MiB request cap. +func mixedSizeOutboxCorpus(seed int64) []outboxRecord { + rng := rand.New(rand.NewSource(seed)) + code := strings.Repeat("if a < b && c > d { return \"x\" }\n", 100) + var records []outboxRecord + add := func(path, content, encoding, kind, target string) { + i := len(records) + records = append(records, outboxRecord{ + CommandID: fmt.Sprintf("mountcmd_%04d", i), + WorkspaceID: "ws_sizing", + RemotePath: path, + ContentType: "text/plain", + Content: content, + Encoding: encoding, + Type: kind, + Target: target, + Mode: uint32(0o644 + (i%2)*0o111), + Hash: fmt.Sprintf("%064x", i), + ExpectedRevision: []string{"0", "rev_12", ""}[i%3], + }) + } + for i := 0; i < 250; i++ { + switch rng.Intn(10) { + case 0: + add(fmt.Sprintf("/repo/empty/%d.txt", i), "", "", "", "") + case 1: + add(fmt.Sprintf("/repo/html/%d.html", i), strings.Repeat("
é", 1+rng.Intn(400)), "", "", "") + case 2: + raw := make([]byte, 1+rng.Intn(64<<10)) + rng.Read(raw) + add(fmt.Sprintf("/repo/bin/%d.bin", i), base64.StdEncoding.EncodeToString(raw), "base64", "", "") + case 3: + add(fmt.Sprintf("/repo/link/%d", i), "../target", "", remoteTypeSymlink, "../target") + case 4: + add(fmt.Sprintf("/slack/channels/C1/messages/factory-create-%d.json", i), `{"text":"draft <&>"}`, "", "", "") + case 5: + add(fmt.Sprintf("/repo/big/%d.txt", i), strings.Repeat("y", 256<<10+rng.Intn(1<<20)), "", "", "") + default: + add(fmt.Sprintf("/repo/src/%d.go", i), code[:rng.Intn(len(code))], "", "", "") + } + } + // Near-cap and over-cap singletons exercise the 8 MiB edge and the raw + // streaming fallback. + add("/repo/huge/near-cap.txt", strings.Repeat("n", int(defaultMaxWritebackBatchBytes)-512), "", "", "") + add("/repo/huge/over-cap.txt", strings.Repeat("o", int(defaultMaxWritebackBatchBytes)+1), "", "", "") + add("/repo/huge/tail.txt", "tail", "", "", "") + return records +} + +var sizingCaps = []int64{0, 1, 300, 4 << 10, 64 << 10, 1 << 20, defaultMaxWritebackBatchBytes} + +func TestBulkWriteRequestSizerMatchesMarshal(t *testing.T) { + records := mixedSizeOutboxCorpus(1) + var sizer bulkWriteRequestSizer + for i, record := range records { + size, ok := bulkWriteFileEncodedSize(outboxRecordAsBulkFile(record)) + sizer.add(size, ok) + if i%37 != 0 && i != len(records)-1 { + continue + } + want := legacyBulkWriteRequestSize(outboxRecordsAsBulkFiles(records[:i+1])) + if got := sizer.size(); got != want { + t.Fatalf("after %d records: incremental size %d, marshal size %d", i+1, got, want) + } + } +} + +func TestChunkOutboxRecordsMatchesLegacyBoundaries(t *testing.T) { + for _, seed := range []int64{1, 2} { + records := mixedSizeOutboxCorpus(seed) + for _, maxBytes := range sizingCaps { + got := chunkOutboxRecords(records, maxBytes) + want := legacyChunkOutboxRecords(records, maxBytes) + if len(got) != len(want) { + t.Fatalf("seed=%d max=%d: %d chunks, legacy %d", seed, maxBytes, len(got), len(want)) + } + for i := range want { + if len(got[i]) != len(want[i]) { + t.Fatalf("seed=%d max=%d chunk %d: %d files, legacy %d", seed, maxBytes, i, len(got[i]), len(want[i])) + } + for j := range want[i] { + if got[i][j].CommandID != want[i][j].CommandID { + t.Fatalf("seed=%d max=%d chunk %d[%d]: %s, legacy %s", seed, maxBytes, i, j, got[i][j].CommandID, want[i][j].CommandID) + } + } + gotBytes := legacyBulkWriteRequestSize(outboxRecordsAsBulkFiles(got[i])) + wantBytes := legacyBulkWriteRequestSize(outboxRecordsAsBulkFiles(want[i])) + if gotBytes != wantBytes { + t.Fatalf("seed=%d max=%d chunk %d: %d request bytes, legacy %d", seed, maxBytes, i, gotBytes, wantBytes) + } + } + } + } +} + +func TestBulkWriteChunkBoundsMatchesLegacyBoundaries(t *testing.T) { + for _, seed := range []int64{1, 2} { + files := outboxRecordsAsBulkFiles(mixedSizeOutboxCorpus(seed)) + for _, maxBytes := range sizingCaps { + got := bulkWriteChunkBounds(files, maxBytes) + want := legacyWriteFilesBulkBounds(files, maxBytes) + if len(got) != len(want) { + t.Fatalf("seed=%d max=%d: %d requests, legacy %d", seed, maxBytes, len(got), len(want)) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("seed=%d max=%d request %d: %+v, legacy %+v", seed, maxBytes, i, got[i], want[i]) + } + } + } + } +} + +func TestChunkPendingBulkWritesMatchesLegacyBoundaries(t *testing.T) { + records := mixedSizeOutboxCorpus(4) + pending := make([]pendingBulkWrite, 0, len(records)) + for _, record := range records { + pending = append(pending, pendingBulkWrite{ + remotePath: record.RemotePath, + snapshot: localSnapshot{ + WireContent: record.Content, + ContentType: record.ContentType, + Encoding: record.Encoding, + Type: record.Type, + Target: record.Target, + Mode: record.Mode, + Hash: record.Hash, + }, + }) + } + for _, maxBytes := range sizingCaps { + got := chunkPendingBulkWrites("ws_sizing", pending, maxBytes) + want := legacyChunkPendingBulkWrites("ws_sizing", pending, maxBytes) + if len(got) != len(want) { + t.Fatalf("max=%d: %d chunks, legacy %d", maxBytes, len(got), len(want)) + } + for i := range want { + if len(got[i]) != len(want[i]) || got[i][0].remotePath != want[i][0].remotePath { + t.Fatalf("max=%d chunk %d differs: %d files from %s, legacy %d from %s", maxBytes, i, len(got[i]), got[i][0].remotePath, len(want[i]), want[i][0].remotePath) + } + } + } +} diff --git a/internal/mountsync/bulk_upload_concurrency_test.go b/internal/mountsync/bulk_upload_concurrency_test.go new file mode 100644 index 00000000..0caad7aa --- /dev/null +++ b/internal/mountsync/bulk_upload_concurrency_test.go @@ -0,0 +1,340 @@ +package mountsync + +import ( + "context" + "fmt" + "net/http" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +// concurrentBulkClient is a goroutine-safe WriteFilesBulk fake. Every other +// RemoteClient method falls through to fakeClient, which the Syncer only calls +// from its lock-holding goroutine. +type concurrentBulkClient struct { + *fakeClient + + mu sync.Mutex + inFlight int + maxInFlight int + uploads map[string]int // remote path -> POSTs that carried it + calls int + // overlap, when > 1, holds each POST until that many are in flight (or a + // short timeout passes) so the test observes real overlap. + overlap int + arrived *sync.Cond + // failFirstPath fails, once, the POST whose first file has this path. + failFirstPath string + failErr error + failed bool +} + +func newConcurrentBulkClient(overlap int) *concurrentBulkClient { + c := &concurrentBulkClient{ + fakeClient: &fakeClient{files: map[string]RemoteFile{}}, + uploads: map[string]int{}, + overlap: overlap, + } + c.arrived = sync.NewCond(&c.mu) + return c +} + +func (c *concurrentBulkClient) WriteFilesBulk(ctx context.Context, workspaceID string, files []BulkWriteFile) (BulkWriteResponse, error) { + c.mu.Lock() + c.calls++ + c.inFlight++ + if c.inFlight > c.maxInFlight { + c.maxInFlight = c.inFlight + } + c.arrived.Broadcast() + deadline := time.Now().Add(200 * time.Millisecond) + for c.overlap > 1 && c.inFlight < c.overlap && time.Now().Before(deadline) { + // sync.Cond has no timed wait; poll briefly while releasing the lock. + c.mu.Unlock() + time.Sleep(time.Millisecond) + c.mu.Lock() + } + fail := !c.failed && c.failFirstPath != "" && files[0].Path == c.failFirstPath + if fail { + c.failed = true + } else { + for _, file := range files { + c.uploads[file.Path]++ + } + } + c.mu.Unlock() + + // Hold briefly so concurrently dispatched chunks really overlap. + time.Sleep(5 * time.Millisecond) + + c.mu.Lock() + c.inFlight-- + c.mu.Unlock() + if fail { + return BulkWriteResponse{}, c.failErr + } + response := BulkWriteResponse{Written: len(files), CorrelationID: "corr"} + for _, file := range files { + response.Results = append(response.Results, BulkWriteResult{Path: file.Path, Revision: "rev_" + strings.TrimPrefix(file.Path, "/repo/")}) + } + return response, nil +} + +// seedConcurrencyOutbox persists n pending records of ~1 KiB in FirstSeenAt +// order and returns their remote paths in that order. +func seedConcurrencyOutbox(t *testing.T, syncer *Syncer, n int) []string { + t.Helper() + base := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) + paths := make([]string, 0, n) + for i := 0; i < n; i++ { + remotePath := fmt.Sprintf("/repo/f%03d.txt", i) + content := fmt.Sprintf("%03d:%s", i, strings.Repeat("x", 1000)) + record := outboxRecord{ + CommandID: fmt.Sprintf("mountcmd_conc_%03d", i), + WorkspaceID: syncer.workspace, + RemotePath: remotePath, + ContentType: "text/plain", + Content: content, + Hash: hashString(content), + Status: outboxStatusPending, + FirstSeenAt: base.Add(time.Duration(i) * time.Second).Format(time.RFC3339Nano), + ExpectedRevision: "0", + } + record.CorrelationID = record.CommandID + if err := syncer.saveOutboxRecord(record); err != nil { + t.Fatalf("seed outbox record: %v", err) + } + paths = append(paths, remotePath) + } + return paths +} + +func newConcurrencySyncer(t *testing.T, client RemoteClient, concurrency int) (*Syncer, string) { + t.Helper() + localDir := t.TempDir() + syncer, err := NewSyncer(client, SyncerOptions{ + WorkspaceID: "ws_bulk_concurrency", + RemoteRoot: "/repo", + LocalRoot: localDir, + }) + if err != nil { + t.Fatalf("NewSyncer: %v", err) + } + syncer.bulkUploadConcurrency = concurrency + return syncer, localDir +} + +func outboxRecordsByPathForTest(t *testing.T, localDir, status string) map[string]outboxRecord { + t.Helper() + out := map[string]outboxRecord{} + for _, record := range readOutboxRecordsInDirForTest(t, filepath.Join(localDir, ".relay", "outbox", status)) { + if _, dup := out[record.RemotePath]; dup { + t.Fatalf("duplicate %s outbox record for %s", status, record.RemotePath) + } + out[record.RemotePath] = record + } + return out +} + +func TestFlushOutboxChunksBoundsInFlightAndAcksEachRecordOnce(t *testing.T) { + // ~1 KiB records under a 4 KiB request cap: three records per chunk, so + // 30 records drain as 10 chunks. + t.Setenv("RELAYFILE_MAX_WRITEBACK_BATCH_BYTES", "4096") + client := newConcurrentBulkClient(3) + syncer, localDir := newConcurrencySyncer(t, client, 3) + paths := seedConcurrencyOutbox(t, syncer, 30) + + if err := syncer.FlushOutboxOnce(context.Background()); err != nil { + t.Fatalf("FlushOutboxOnce: %v", err) + } + + if client.calls != 10 { + t.Fatalf("bulk POSTs = %d, want 10 (chunk boundaries changed)", client.calls) + } + if client.maxInFlight != 3 { + t.Fatalf("max in-flight POSTs = %d, want exactly the configured 3", client.maxInFlight) + } + acked := outboxRecordsByPathForTest(t, localDir, "acked") + pending := outboxRecordsByPathForTest(t, localDir, "pending") + if len(pending) != 0 { + t.Fatalf("pending after flush = %d, want 0", len(pending)) + } + for _, path := range paths { + if client.uploads[path] != 1 { + t.Fatalf("%s uploaded %d times, want 1", path, client.uploads[path]) + } + if _, ok := acked[path]; !ok { + t.Fatalf("%s not acked", path) + } + if got := syncer.state.Files[path].Revision; got != "rev_"+strings.TrimPrefix(path, "/repo/") { + t.Fatalf("%s tracked revision = %q", path, got) + } + } + if len(acked) != len(paths) { + t.Fatalf("acked = %d, want %d", len(acked), len(paths)) + } +} + +func TestFlushOutboxChunksSerialWhenConcurrencyIsOne(t *testing.T) { + t.Setenv("RELAYFILE_MAX_WRITEBACK_BATCH_BYTES", "4096") + client := newConcurrentBulkClient(1) + syncer, _ := newConcurrencySyncer(t, client, 1) + seedConcurrencyOutbox(t, syncer, 12) + if err := syncer.FlushOutboxOnce(context.Background()); err != nil { + t.Fatalf("FlushOutboxOnce: %v", err) + } + if client.maxInFlight != 1 || client.calls != 4 { + t.Fatalf("serial flush: maxInFlight=%d calls=%d, want 1 and 4", client.maxInFlight, client.calls) + } +} + +// A 429 workspace_busy that outlives the client's own Retry-After retries +// fails only its chunk: those records stay pending with backoff, chunks +// already in flight are still acked (never re-sent), no later chunk is +// dispatched, and the next flush delivers the rest exactly once. +func TestFlushOutboxChunksWorkspaceBusyKeepsAccountingExact(t *testing.T) { + t.Setenv("RELAYFILE_MAX_WRITEBACK_BATCH_BYTES", "4096") + client := newConcurrentBulkClient(2) + syncer, localDir := newConcurrencySyncer(t, client, 2) + paths := seedConcurrencyOutbox(t, syncer, 30) + // Chunks are [0-2] [3-5] [6-8] ...; fail the second chunk. + client.failFirstPath = paths[3] + client.failErr = &HTTPError{StatusCode: http.StatusTooManyRequests, Code: "workspace_busy", Message: "workspace busy"} + + err := syncer.FlushOutboxOnce(context.Background()) + if err == nil { + t.Fatalf("FlushOutboxOnce succeeded despite a workspace_busy chunk (failed=%v calls=%d)", client.failed, client.calls) + } + + acked := outboxRecordsByPathForTest(t, localDir, "acked") + pending := outboxRecordsByPathForTest(t, localDir, "pending") + for path := range acked { + if _, both := pending[path]; both { + t.Fatalf("%s is both acked and pending", path) + } + } + if len(acked)+len(pending) != len(paths) { + t.Fatalf("acked %d + pending %d != %d records", len(acked), len(pending), len(paths)) + } + for _, path := range paths[:3] { + if _, ok := acked[path]; !ok { + t.Fatalf("%s from the chunk before the failure was not acked", path) + } + } + for _, path := range paths[3:6] { + record, ok := pending[path] + if !ok { + t.Fatalf("%s from the failed chunk is not pending", path) + } + if record.AttemptCount != 1 || record.NextAttemptAt == "" || !strings.Contains(record.LastError, "busy") { + t.Fatalf("%s failed-chunk record = attempts %d next %q err %q, want one backed-off attempt", path, record.AttemptCount, record.NextAttemptAt, record.LastError) + } + } + // With two in flight, the third chunk was dispatched alongside the failing + // second one and the server accepted it: it must be acked, not resent. + for _, path := range paths[6:9] { + if _, ok := acked[path]; !ok { + t.Fatalf("%s was accepted by the server while the failed chunk was in flight but not acked", path) + } + } + for path := range pending { + if client.uploads[path] != 0 { + t.Fatalf("pending %s was already accepted by the server %d time(s)", path, client.uploads[path]) + } + } + if client.calls >= 10 { + t.Fatalf("chunks after the failure were still dispatched: %d POSTs", client.calls) + } + + // The next drain (FlushOutboxOnce forces due) delivers everything left. + if err := syncer.FlushOutboxOnce(context.Background()); err != nil { + t.Fatalf("second FlushOutboxOnce: %v", err) + } + acked = outboxRecordsByPathForTest(t, localDir, "acked") + if pending := outboxRecordsByPathForTest(t, localDir, "pending"); len(pending) != 0 { + t.Fatalf("pending after recovery = %d", len(pending)) + } + for _, path := range paths { + if client.uploads[path] != 1 { + t.Fatalf("%s accepted %d times across both flushes, want exactly 1", path, client.uploads[path]) + } + if _, ok := acked[path]; !ok { + t.Fatalf("%s not acked after recovery", path) + } + } + if client.maxInFlight > 2 { + t.Fatalf("max in-flight = %d exceeds the configured 2", client.maxInFlight) + } +} + +// Two pending commands for one path must never be in flight together. +func TestFlushOutboxChunksSerializesSamePathAcrossChunks(t *testing.T) { + client := newConcurrentBulkClient(1) + var order []string + var orderMu sync.Mutex + syncer, _ := newConcurrencySyncer(t, &pathOrderClient{concurrentBulkClient: client, order: &order, mu: &orderMu}, 3) + var chunks [][]outboxRecord + for i, remotePath := range []string{"/repo/a.txt", "/repo/b.txt", "/repo/a.txt"} { + record := outboxRecord{ + CommandID: fmt.Sprintf("mountcmd_samepath_%d", i), + WorkspaceID: syncer.workspace, + RemotePath: remotePath, + ContentType: "text/plain", + Content: fmt.Sprintf("v%d", i), + Hash: hashString(fmt.Sprintf("v%d", i)), + Status: outboxStatusPending, + FirstSeenAt: time.Now().UTC().Format(time.RFC3339Nano), + } + if err := syncer.saveOutboxRecord(record); err != nil { + t.Fatalf("seed: %v", err) + } + chunks = append(chunks, []outboxRecord{record}) + } + syncer.mu.Lock() + err := syncer.flushOutboxChunks(context.Background(), chunks, map[string]struct{}{}, true, false) + syncer.mu.Unlock() + if err != nil { + t.Fatalf("flushOutboxChunks: %v", err) + } + if client.maxInFlight > 2 { + t.Fatalf("max in flight = %d, want <= 2 (a.txt twice must not overlap)", client.maxInFlight) + } + orderMu.Lock() + defer orderMu.Unlock() + firstA, secondA := -1, -1 + for i, event := range order { + switch event { + case "end:/repo/a.txt": + if firstA < 0 { + firstA = i + } + case "start:/repo/a.txt": + if firstA >= 0 { + secondA = i + } + } + } + if firstA < 0 || secondA < 0 || secondA < firstA { + t.Fatalf("second a.txt upload started before the first finished: %v", order) + } +} + +type pathOrderClient struct { + *concurrentBulkClient + order *[]string + mu *sync.Mutex +} + +func (c *pathOrderClient) WriteFilesBulk(ctx context.Context, workspaceID string, files []BulkWriteFile) (BulkWriteResponse, error) { + c.mu.Lock() + *c.order = append(*c.order, "start:"+files[0].Path) + c.mu.Unlock() + response, err := c.concurrentBulkClient.WriteFilesBulk(ctx, workspaceID, files) + c.mu.Lock() + *c.order = append(*c.order, "end:"+files[0].Path) + c.mu.Unlock() + return response, err +} diff --git a/internal/mountsync/push_local_read_once_test.go b/internal/mountsync/push_local_read_once_test.go new file mode 100644 index 00000000..743ef1f5 --- /dev/null +++ b/internal/mountsync/push_local_read_once_test.go @@ -0,0 +1,144 @@ +package mountsync + +import ( + "bytes" + "context" + "os" + "path/filepath" + "sync" + "testing" +) + +type snapshotReadCounter struct { + mu sync.Mutex + reads map[string]int + full map[string]int +} + +func countSnapshotReads(syncer *Syncer) *snapshotReadCounter { + counter := &snapshotReadCounter{reads: map[string]int{}, full: map[string]int{}} + syncer.readLocalSnapshotFn = func(path string, includeContent bool) (localSnapshot, error) { + counter.mu.Lock() + counter.reads[filepath.Base(path)]++ + if includeContent { + counter.full[filepath.Base(path)]++ + } + counter.mu.Unlock() + return readLocalSnapshotLimitedUnderRoot(syncer.localRoot, path, includeContent, maxWritebackBytes()) + } + return counter +} + +// A new (untracked) file used to be read three times by +// PushLocalAndFlushOnce: hashed by pushLocal's scan, re-read whole to build +// the upload, then hashed again by saveState's public-state scan. It is now +// read exactly once, and the uploaded bytes are the bytes on disk. +func TestPushLocalAndFlushOnceReadsNewFileOnce(t *testing.T) { + t.Setenv("RELAYFILE_MAX_WRITEBACK_BYTES", "4096") + localDir := t.TempDir() + newBody := []byte("package main\n\nfunc main() { println(\"<&>\") }\n") + if err := os.WriteFile(filepath.Join(localDir, "new.go"), newBody, 0o644); err != nil { + t.Fatal(err) + } + // Oversized files are never read whole and must still surface as + // writeback-skipped in the public state derived from the reused scan. + if err := os.WriteFile(filepath.Join(localDir, "huge.bin"), bytes.Repeat([]byte("h"), 8192), 0o644); err != nil { + t.Fatal(err) + } + client := &fakeClient{files: map[string]RemoteFile{}} + syncer, err := NewSyncer(client, SyncerOptions{ + WorkspaceID: "ws_read_once", + RemoteRoot: "/", + LocalRoot: localDir, + }) + if err != nil { + t.Fatalf("NewSyncer: %v", err) + } + counter := countSnapshotReads(syncer) + + if err := syncer.PushLocalAndFlushOnce(context.Background()); err != nil { + t.Fatalf("PushLocalAndFlushOnce: %v", err) + } + + if got := counter.reads["new.go"]; got != 1 { + t.Fatalf("new.go read %d times, want 1", got) + } + if got := counter.full["new.go"]; got != 1 { + t.Fatalf("new.go full reads = %d, want 1", got) + } + if got := counter.reads["huge.bin"]; got != 0 { + t.Fatalf("oversized huge.bin read %d times, want 0", got) + } + if client.bulkWriteCalls != 1 || len(client.bulkWriteBatches[0]) != 1 { + t.Fatalf("bulk writes = %d %+v, want one batch with new.go", client.bulkWriteCalls, client.bulkWriteBatches) + } + uploaded := client.bulkWriteBatches[0][0] + if uploaded.Path != "/new.go" || uploaded.Content != string(newBody) { + t.Fatalf("uploaded %s %q, want /new.go with on-disk bytes", uploaded.Path, uploaded.Content) + } + if got := syncer.state.Files["/new.go"].Hash; got != hashBytes(newBody) { + t.Fatalf("tracked hash = %s, want hash of on-disk bytes", got) + } + state := readPublicState(t, localDir) + if got := state.Files["/huge.bin"].Status; got != "writeback-skipped" { + t.Fatalf("huge.bin public status = %q, want writeback-skipped", got) + } + if got := state.Files["/new.go"].Status; got != "ready" { + t.Fatalf("new.go public status = %q, want ready", got) + } +} + +// Tracked dirty files are not captured by the scan (their hash decides +// whether they changed at all), so they keep the hash-then-read path. +func TestPushLocalCapturesOnlyUntrackedFiles(t *testing.T) { + localDir := t.TempDir() + if err := os.WriteFile(filepath.Join(localDir, "tracked.txt"), []byte("edited\n"), 0o644); err != nil { + t.Fatal(err) + } + client := &fakeClient{files: map[string]RemoteFile{}} + syncer, err := NewSyncer(client, SyncerOptions{ + WorkspaceID: "ws_capture_untracked", + RemoteRoot: "/", + LocalRoot: localDir, + }) + if err != nil { + t.Fatalf("NewSyncer: %v", err) + } + syncer.state.Files["/tracked.txt"] = trackedFile{Revision: "rev_1", Hash: hashBytes([]byte("base\n")), Type: remoteTypeFile, Mode: 0o644} + counter := countSnapshotReads(syncer) + + if _, err := syncer.pushLocal(context.Background()); err != nil { + t.Fatalf("pushLocal: %v", err) + } + if got, full := counter.reads["tracked.txt"], counter.full["tracked.txt"]; got != 2 || full != 1 { + t.Fatalf("tracked.txt reads=%d full=%d, want hash scan + one full read", got, full) + } + if client.bulkWriteCalls != 1 || client.bulkWriteBatches[0][0].Content != "edited\n" { + t.Fatalf("tracked edit not uploaded: %+v", client.bulkWriteBatches) + } +} + +// In low-memory mode the scan never retains file content. +func TestPushLocalLowMemoryDoesNotCaptureContent(t *testing.T) { + localDir := t.TempDir() + if err := os.WriteFile(filepath.Join(localDir, "new.txt"), []byte("hello\n"), 0o644); err != nil { + t.Fatal(err) + } + lowMemory := true + syncer, err := NewSyncer(&fakeClient{files: map[string]RemoteFile{}}, SyncerOptions{ + WorkspaceID: "ws_capture_low_memory", + RemoteRoot: "/", + LocalRoot: localDir, + LowMemory: &lowMemory, + }) + if err != nil { + t.Fatalf("NewSyncer: %v", err) + } + _, captured, err := syncer.scanLocalFilesCapturing(true) + if err != nil { + t.Fatalf("scan: %v", err) + } + if len(captured) != 0 { + t.Fatalf("low-memory scan captured %d files", len(captured)) + } +} diff --git a/internal/mountsync/push_perf_test.go b/internal/mountsync/push_perf_test.go new file mode 100644 index 00000000..906eb9a9 --- /dev/null +++ b/internal/mountsync/push_perf_test.go @@ -0,0 +1,216 @@ +package mountsync + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/agentworkforce/relayfile/internal/relayfile" +) + +// pushPerfStats observes the /fs/bulk traffic a push produces so the large +// corpus benchmark can report how many requests went out, how long the +// client stayed silent before the first one, and how many overlapped. +type pushPerfStats struct { + started time.Time + bulkPosts atomic.Int64 + requests atomic.Int64 + inFlight atomic.Int64 + maxInFlight atomic.Int64 + firstPostMu sync.Mutex + firstPost time.Duration +} + +func (s *pushPerfStats) wrap(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + s.requests.Add(1) + if r.Method == http.MethodPost && strings.HasSuffix(r.URL.Path, "/fs/bulk") { + s.firstPostMu.Lock() + if s.firstPost == 0 { + s.firstPost = time.Since(s.started) + } + s.firstPostMu.Unlock() + s.bulkPosts.Add(1) + current := s.inFlight.Add(1) + for { + observed := s.maxInFlight.Load() + if current <= observed || s.maxInFlight.CompareAndSwap(observed, current) { + break + } + } + defer s.inFlight.Add(-1) + } + next.ServeHTTP(w, r) + }) +} + +// seedPushPerfCorpus writes n deterministic text files of roughly size bytes +// spread across nested directories, approximating a seeded code tree. +func seedPushPerfCorpus(tb testing.TB, root string, n, size int) { + tb.Helper() + line := "func example() { return \"relayfile mount push benchmark payload\" }\n" + body := strings.Repeat(line, size/len(line)+1)[:size] + for i := 0; i < n; i++ { + dir := filepath.Join(root, "src", fmt.Sprintf("pkg%03d", i/100), fmt.Sprintf("sub%d", i%7)) + if err := os.MkdirAll(dir, 0o755); err != nil { + tb.Fatal(err) + } + content := fmt.Sprintf("// file %d\n%s", i, body) + if err := os.WriteFile(filepath.Join(dir, fmt.Sprintf("file%05d.go", i)), []byte(content), 0o644); err != nil { + tb.Fatal(err) + } + } +} + +type pushPerfRun struct { + syncer *Syncer + stats *pushPerfStats + close func() +} + +func newPushPerfRun(tb testing.TB, n, size int) pushPerfRun { + tb.Helper() + // The reference store's default 1024-slot writeback queue drops (and + // leaves pending) receipts when ~2.7k-file chunks land back to back; size + // it for the corpus so receipts settle and only client cost is measured. + store := relayfile.NewStoreWithOptions(relayfile.StoreOptions{WritebackQueue: relayfile.NewInMemoryWritebackQueue(1 << 16)}) + tb.Cleanup(store.Close) + stats := &pushPerfStats{} + api := httptest.NewServer(stats.wrap(newMountsyncAPIHandler(tb, store))) + workspaceID := "ws_push_perf" + token := mustMountsyncTestJWT(tb, "dev-secret", workspaceID, "push-perf", []string{"fs:read", "fs:write", "sync:trigger", "ops:read"}, time.Now().Add(time.Hour)) + client := NewHTTPClient(api.URL, token, api.Client()) + localRoot := tb.TempDir() + seedPushPerfCorpus(tb, localRoot, n, size) + syncer, err := NewSyncer(client, SyncerOptions{ + WorkspaceID: workspaceID, + RemoteRoot: "/repo", + LocalRoot: localRoot, + StateDir: tb.TempDir(), + RootCtx: context.Background(), + FullPullEvery: -1, + WebSocket: boolPtr(false), + }) + if err != nil { + tb.Fatalf("NewSyncer: %v", err) + } + return pushPerfRun{syncer: syncer, stats: stats, close: api.Close} +} + +// settleLaggingReceipts tolerates exactly one benign outcome of a drain +// against the in-process reference server: every write was accepted, but the +// server's asynchronous writeback dispatch had not yet marked a few operation +// receipts succeeded when the client polled them. Those records keep their +// opId and zero failed attempts; a later receipt poll settles them without +// re-uploading. Anything else fails the benchmark. +func settleLaggingReceipts(b *testing.B, syncer *Syncer, err error) { + b.Helper() + if err == nil { + return + } + records, listErr := syncer.listPendingOutboxRecords() + if listErr != nil || len(records) == 0 { + b.Fatalf("drain failed: %v", err) + } + for _, record := range records { + if record.OpID == "" || record.AttemptCount != 0 || record.NeedsAttention { + b.Fatalf("drain failed: %v (record %s attempts=%d opId=%q err=%q)", err, record.RemotePath, record.AttemptCount, record.OpID, record.LastError) + } + } + b.Logf("%d receipt(s) still pending server dispatch; polling again", len(records)) + time.Sleep(2 * time.Second) + if err := syncer.FlushOutboxOnce(context.Background()); err != nil { + b.Fatalf("receipt re-poll failed: %v", err) + } +} + +// BenchmarkPushLocalAndFlushOnceLargeTree reproduces the production +// `relayfile-mount --push-local-once` seed of a ~13k file code tree against +// the real HTTP API handler. Run with: +// +// go test ./internal/mountsync -run '^$' -bench PushLocalAndFlushOnceLargeTree -benchtime 1x -benchmem +func BenchmarkPushLocalAndFlushOnceLargeTree(b *testing.B) { + const files, size = 13000, 3 * 1024 + for i := 0; i < b.N; i++ { + b.StopTimer() + run := newPushPerfRun(b, files, size) + run.stats.started = time.Now() + b.StartTimer() + err := run.syncer.PushLocalAndFlushOnce(context.Background()) + b.StopTimer() + b.ReportMetric(float64(run.stats.bulkPosts.Load()), "bulk_posts/op") + b.ReportMetric(float64(run.stats.firstPost.Milliseconds()), "first_post_ms/op") + b.ReportMetric(float64(run.stats.maxInFlight.Load()), "max_in_flight") + b.ReportMetric(float64(run.stats.requests.Load()), "http_requests/op") + settleLaggingReceipts(b, run.syncer, err) + if outbox := run.syncer.summarizeOutbox(); outbox.Pending != 0 || outbox.NeedsAttention != 0 { + b.Fatalf("outbox not drained: %+v", outbox) + } + run.close() + b.StartTimer() + } +} + +// seedPushPerfOutbox ingests every corpus file into the durable outbox +// without uploading, reproducing the backlog a later flush drains in +// maxWritebackBatchBytes chunks (the production pattern of a handful of +// strictly serial 8 MiB /fs/bulk POSTs). +func seedPushPerfOutbox(tb testing.TB, syncer *Syncer) int { + tb.Helper() + localFiles, err := syncer.scanLocalFiles() + if err != nil { + tb.Fatal(err) + } + count := 0 + for remotePath, snapshot := range localFiles { + full, err := syncer.readLocalSnapshot(snapshot.LocalPath, true) + if err != nil { + tb.Fatal(err) + } + if _, err := syncer.ensureOutboxRecord(pendingBulkWrite{remotePath: remotePath, localPath: snapshot.LocalPath, snapshot: full}); err != nil { + tb.Fatal(err) + } + count++ + } + return count +} + +// BenchmarkFlushOutboxOnceLargeBacklog drains a ~13k record (~41 MB) outbox +// backlog. Run with: +// +// go test ./internal/mountsync -run '^$' -bench FlushOutboxOnceLargeBacklog -benchtime 1x -benchmem +func BenchmarkFlushOutboxOnceLargeBacklog(b *testing.B) { + const files, size = 13000, 3 * 1024 + for i := 0; i < b.N; i++ { + b.StopTimer() + run := newPushPerfRun(b, files, size) + if got := seedPushPerfOutbox(b, run.syncer); got != files { + b.Fatalf("seeded %d records, want %d", got, files) + } + // Measure the drain itself. Before linear batch sizing this drain + // overran the default 60s outbox flush budget on an Apple M3. + run.syncer.outboxFlushTimeout = time.Hour + run.stats.started = time.Now() + b.StartTimer() + err := run.syncer.FlushOutboxOnce(context.Background()) + b.StopTimer() + b.ReportMetric(float64(run.stats.bulkPosts.Load()), "bulk_posts/op") + b.ReportMetric(float64(run.stats.firstPost.Milliseconds()), "first_post_ms/op") + b.ReportMetric(float64(run.stats.maxInFlight.Load()), "max_in_flight") + b.ReportMetric(float64(run.stats.requests.Load()), "http_requests/op") + settleLaggingReceipts(b, run.syncer, err) + if outbox := run.syncer.summarizeOutbox(); outbox.Pending != 0 || outbox.NeedsAttention != 0 { + b.Fatalf("outbox not drained: %+v", outbox) + } + run.close() + b.StartTimer() + } +} diff --git a/internal/mountsync/stream_write.go b/internal/mountsync/stream_write.go index 5d5d1a52..5296537b 100644 --- a/internal/mountsync/stream_write.go +++ b/internal/mountsync/stream_write.go @@ -18,15 +18,11 @@ func (c *HTTPClient) WriteFilesBulk(ctx context.Context, workspaceID string, fil return BulkWriteResponse{}, ErrEmptyBulkWrite } var aggregate BulkWriteResponse - for start := 0; start < len(files); { - end := start + 1 - for end < len(files) && bulkWriteRequestSize(files[start:end+1]) <= defaultMaxWritebackBatchBytes { - end++ - } - chunk := files[start:end] + for _, bounds := range bulkWriteChunkBounds(files, defaultMaxWritebackBatchBytes) { + chunk := files[bounds.start:bounds.end] var out BulkWriteResponse var err error - if len(chunk) == 1 && bulkWriteRequestSize(chunk) > defaultMaxWritebackBatchBytes { + if len(chunk) == 1 && bounds.size > defaultMaxWritebackBatchBytes { out, err = c.writeLargeFile(ctx, workspaceID, chunk[0]) } else { out, err = c.writeFilesBulkJSON(ctx, workspaceID, chunk) @@ -39,11 +35,44 @@ func (c *HTTPClient) WriteFilesBulk(ctx context.Context, workspaceID string, fil aggregate.Errors = append(aggregate.Errors, out.Errors...) aggregate.Results = append(aggregate.Results, out.Results...) aggregate.CorrelationID = out.CorrelationID - start = end } return aggregate, nil } +type bulkWriteChunkBound struct { + start, end int + // size is bulkWriteRequestSize(files[start:end]). + size int64 +} + +// bulkWriteChunkBounds greedily packs files into JSON requests of at most +// maxBytes, always taking at least one file per request. Each file is encoded +// once; the boundaries match re-measuring bulkWriteRequestSize on every +// append (TestBulkWriteChunkBoundsMatchesLegacyBoundaries). +func bulkWriteChunkBounds(files []BulkWriteFile, maxBytes int64) []bulkWriteChunkBound { + type encoded struct { + size int64 + ok bool + } + sizes := make([]encoded, len(files)) + for i := range files { + sizes[i].size, sizes[i].ok = bulkWriteFileEncodedSize(files[i]) + } + bounds := make([]bulkWriteChunkBound, 0, 1) + for start := 0; start < len(files); { + var sizer bulkWriteRequestSizer + sizer.add(sizes[start].size, sizes[start].ok) + end := start + 1 + for end < len(files) && sizer.sizeWith(sizes[end].size, sizes[end].ok) <= maxBytes { + sizer.add(sizes[end].size, sizes[end].ok) + end++ + } + bounds = append(bounds, bulkWriteChunkBound{start: start, end: end, size: sizer.size()}) + start = end + } + return bounds +} + func (c *HTTPClient) writeLargeFile(ctx context.Context, workspaceID string, file BulkWriteFile) (BulkWriteResponse, error) { if isSymlinkType(file.Type) || file.WritebackIntent != "" { return c.writeFilesBulkJSON(ctx, workspaceID, []BulkWriteFile{file}) diff --git a/internal/mountsync/syncer.go b/internal/mountsync/syncer.go index b8666edd..3a91d856 100644 --- a/internal/mountsync/syncer.go +++ b/internal/mountsync/syncer.go @@ -1877,6 +1877,7 @@ type Syncer struct { wsLastPingAt time.Time listenerHeartbeatAt time.Time bulkFlushThreshold int + bulkUploadConcurrency int mode string interval time.Duration fullPullEvery int @@ -1931,6 +1932,13 @@ type Syncer struct { circuit *CloudErrorCircuit maxOutboxAttempts int nowFn func() time.Time + + // retainPushLocalScan asks pushLocal to keep its hash-only scan in + // lastPushLocalScan so a one-shot caller can derive the public state from + // it instead of re-hashing the whole tree (PushLocalAndFlushOnce). + retainPushLocalScan bool + lastPushLocalScan map[string]localSnapshot + // Test seam for proving that background local-tree hashing yields the // real-time state lock. Production uses readLocalSnapshot directly. readLocalSnapshotFn func(string, bool) (localSnapshot, error) @@ -2723,6 +2731,7 @@ func NewSyncer(client RemoteClient, opts SyncerOptions) (*Syncer, error) { logger: opts.Logger, denialLogPath: filepath.Join(localRoot, ".relay", "permissions-denied.log"), bulkFlushThreshold: bulkFlushThreshold, + bulkUploadConcurrency: bulkUploadConcurrencyFromEnv(opts.Logger), mode: strings.TrimSpace(opts.Mode), pullOnly: normalizeSyncMode(opts.SyncMode) == "pull-only", writeOnly: normalizeSyncMode(opts.SyncMode) == "write-only", @@ -2951,6 +2960,12 @@ func (s *Syncer) PushLocalAndFlushOnce(ctx context.Context) error { if err := s.loadState(); err != nil { return err } + s.retainPushLocalScan = true + s.lastPushLocalScan = nil + defer func() { + s.retainPushLocalScan = false + s.lastPushLocalScan = nil + }() conflicted, err := s.pushLocal(ctx) if err != nil { s.markSyncError(err) @@ -2976,6 +2991,14 @@ func (s *Syncer) PushLocalAndFlushOnce(ctx context.Context) error { return err } s.markSyncSuccess() + if s.lastPushLocalScan != nil { + // pushLocal just walked and hashed the tree. The flush that followed + // only uploads; its conflict/read-only handling never changes which + // paths exist locally or whether they are writeback-skipped — the + // only facts the public state reads from a scan. Reuse that scan + // rather than hashing every file a third time. + return s.saveStateWithLocalFiles(s.lastPushLocalScan) + } return s.saveState() } @@ -4102,12 +4125,7 @@ func (s *Syncer) flushSelectedOutboxRecords(ctx context.Context, records []outbo } flushCtx, cancel := s.outboxContext(ctx) defer cancel() - for _, chunk := range chunkOutboxRecords(due, maxWritebackBatchBytes()) { - if err := s.flushOutboxRecordChunk(flushCtx, chunk, conflicted, settleReceipts, unlockDuringUpload); err != nil { - return err - } - } - return nil + return s.flushOutboxChunks(flushCtx, chunkOutboxRecords(due, maxWritebackBatchBytes()), conflicted, settleReceipts, unlockDuringUpload) } func (s *Syncer) flushDueOutboxRecords(ctx context.Context, conflicted map[string]struct{}) error { @@ -4155,19 +4173,146 @@ func (s *Syncer) flushOutboxRecords(ctx context.Context, conflicted map[string]s // writeback mid-flight and leave it retrying for minutes (see outboxContext). flushCtx, cancel := s.outboxContext(ctx) defer cancel() - for _, chunk := range chunkOutboxRecords(due, maxWritebackBatchBytes()) { - if err := s.flushOutboxRecordChunk(flushCtx, chunk, conflicted, settleReceipts, false); err != nil { - return err + return s.flushOutboxChunks(flushCtx, chunkOutboxRecords(due, maxWritebackBatchBytes()), conflicted, settleReceipts, false) +} + +// flushOutboxChunks flushes chunks in order. With a single chunk, a +// concurrency of 1, or unlockDuringUpload (the watcher path, whose state lock +// handoff is per request), it is the original serial loop. +// +// Otherwise it pipelines up to bulkUploadConcurrency POSTs while keeping +// every state transition serial and in chunk order on the calling goroutine, +// which holds s.mu throughout: +// +// - prepare (receipt settlement, capability gate) runs under s.mu, in order; +// - only the immutable request upload runs on a worker goroutine; +// - responses are applied under s.mu strictly in chunk order, so acks, +// failures, backoff and conflict handling see the same sequence as the +// serial loop, and a record is acked only after the server confirmed it; +// - a chunk sharing any path with an in-flight chunk waits for it, so two +// commands for one path are never in flight together; +// - after the first error no further chunk is dispatched (as before), but +// uploads already in flight are still applied, so a write the server +// accepted is acked instead of being re-sent. +func (s *Syncer) flushOutboxChunks(ctx context.Context, chunks [][]outboxRecord, conflicted map[string]struct{}, settleReceipts, unlockDuringUpload bool) error { + limit := s.bulkUploadConcurrencyValue() + if len(chunks) <= 1 || limit <= 1 || unlockDuringUpload { + for _, chunk := range chunks { + if err := s.flushOutboxRecordChunk(ctx, chunk, conflicted, settleReceipts, unlockDuringUpload); err != nil { + return err + } } + return nil } - return nil + + type uploadResult struct { + response BulkWriteResponse + err error + } + type inflightChunk struct { + upload outboxChunkUpload + paths []string + started time.Time + done chan uploadResult + } + var ( + queue []*inflightChunk + inflightBy = map[string]int{} + next int + firstErr error + stopDispatch bool + ) + conflictsInflight := func(chunk []outboxRecord) bool { + for _, record := range chunk { + if inflightBy[normalizeRemotePath(record.RemotePath)] > 0 { + return true + } + } + return false + } + for { + for !stopDispatch && next < len(chunks) && len(queue) < limit && !conflictsInflight(chunks[next]) { + chunk := chunks[next] + next++ + upload, done, err := s.prepareOutboxChunkUpload(ctx, chunk, settleReceipts) + if done { + if err != nil { + firstErr = err + stopDispatch = true + } + continue + } + item := &inflightChunk{upload: upload, started: time.Now(), done: make(chan uploadResult, 1)} + for _, record := range upload.records { + path := normalizeRemotePath(record.RemotePath) + item.paths = append(item.paths, path) + inflightBy[path]++ + } + queue = append(queue, item) + go func(files []BulkWriteFile, done chan<- uploadResult) { + response, err := s.client.WriteFilesBulk(ctx, s.workspace, files) + done <- uploadResult{response: response, err: err} + }(upload.files, item.done) + } + if len(queue) == 0 { + break + } + head := queue[0] + queue = queue[1:] + result := <-head.done + for _, path := range head.paths { + if inflightBy[path]--; inflightBy[path] <= 0 { + delete(inflightBy, path) + } + } + s.traceLatency("bulk_response", head.started, "records", len(head.upload.records)) + if err := s.applyOutboxChunkResponse(ctx, head.upload, result.response, result.err, conflicted, settleReceipts); err != nil && firstErr == nil { + firstErr = err + stopDispatch = true + } + } + return firstErr } func (s *Syncer) flushOutboxRecordChunk(ctx context.Context, records []outboxRecord, conflicted map[string]struct{}, settleReceipts, unlockDuringUpload bool) error { traceStarted := time.Now() + upload, done, err := s.prepareOutboxChunkUpload(ctx, records, settleReceipts) + if done { + return err + } + var response BulkWriteResponse + if unlockDuringUpload { + // The watcher has already persisted immutable outbox records, so the + // network wait no longer needs exclusive access to mount state. Let the + // WebSocket apply loop materialize peer edits while our own POST is in + // flight; reacquire before interpreting the response or mutating state. + s.mu.Unlock() + response, err = s.client.WriteFilesBulk(ctx, s.workspace, upload.files) + s.mu.Lock() + } else { + response, err = s.client.WriteFilesBulk(ctx, s.workspace, upload.files) + } + s.traceLatency("bulk_response", traceStarted, "records", len(upload.records)) + return s.applyOutboxChunkResponse(ctx, upload, response, err, conflicted, settleReceipts) +} + +// outboxChunkUpload is the network-only part of one outbox chunk flush: the +// records that still need a POST and their immutable request files. The +// upload itself touches no Syncer state, so it may run without s.mu and +// concurrently with other chunks' uploads (see flushOutboxChunks). +type outboxChunkUpload struct { + records []outboxRecord + files []BulkWriteFile + firstErr error +} + +// prepareOutboxChunkUpload runs the state-touching pre-upload half of a chunk +// flush under s.mu: control-path skips, receipt settlement for records that +// already hold an opId, and the symlink capability gate. done reports that +// nothing remains to upload, in which case err is the chunk's result. +func (s *Syncer) prepareOutboxChunkUpload(ctx context.Context, records []outboxRecord, settleReceipts bool) (outboxChunkUpload, bool, error) { var firstErr error uploadRecords := make([]outboxRecord, 0, len(records)) - deferredReceipts := make([]outboxRecord, 0, len(records)) for _, record := range records { if isMountRuntimeRemotePath(record.RemotePath) { s.logMountControlPathSkipped(strings.TrimPrefix(normalizeRemotePath(record.RemotePath), "/")) @@ -4187,14 +4332,14 @@ func (s *Syncer) flushOutboxRecordChunk(ctx context.Context, records []outboxRec uploadRecords = append(uploadRecords, record) } if len(uploadRecords) == 0 { - return firstErr + return outboxChunkUpload{}, true, firstErr } files := outboxRecordsAsBulkFiles(uploadRecords) if hasSymlinkBulkWrite(files) { capabilityClient, ok := s.client.(symlinkCapabilityClient) if !ok { - return fmt.Errorf("%w: client did not advertise a capability check", ErrSymlinkUnsupported) + return outboxChunkUpload{}, true, fmt.Errorf("%w: client did not advertise a capability check", ErrSymlinkUnsupported) } if err := capabilityClient.EnsureSymlinkSupport(ctx); err != nil { for _, record := range uploadRecords { @@ -4203,25 +4348,22 @@ func (s *Syncer) flushOutboxRecordChunk(ctx context.Context, records []outboxRec } } if firstErr != nil { - return firstErr + return outboxChunkUpload{}, true, firstErr } - return err + return outboxChunkUpload{}, true, err } } - var response BulkWriteResponse - var err error - if unlockDuringUpload { - // The watcher has already persisted immutable outbox records, so the - // network wait no longer needs exclusive access to mount state. Let the - // WebSocket apply loop materialize peer edits while our own POST is in - // flight; reacquire before interpreting the response or mutating state. - s.mu.Unlock() - response, err = s.client.WriteFilesBulk(ctx, s.workspace, files) - s.mu.Lock() - } else { - response, err = s.client.WriteFilesBulk(ctx, s.workspace, files) - } - s.traceLatency("bulk_response", traceStarted, "records", len(uploadRecords)) + return outboxChunkUpload{records: uploadRecords, files: files, firstErr: firstErr}, false, nil +} + +// applyOutboxChunkResponse is the post-upload half of a chunk flush, run under +// s.mu: it records the cloud outcome and acks, fails, or schedules a retry +// for every uploaded record. Records are acknowledged only from a server +// response; a transport error leaves them pending with backoff. +func (s *Syncer) applyOutboxChunkResponse(ctx context.Context, upload outboxChunkUpload, response BulkWriteResponse, err error, conflicted map[string]struct{}, settleReceipts bool) error { + firstErr := upload.firstErr + uploadRecords := upload.records + deferredReceipts := make([]outboxRecord, 0, len(uploadRecords)) if err != nil { s.recordCloudFailure(err) for _, record := range uploadRecords { @@ -4551,15 +4693,24 @@ func chunkOutboxRecords(records []outboxRecord, maxBytes int64) [][]outboxRecord if maxBytes <= 0 { return [][]outboxRecord{records} } + // Each record is encoded exactly once and the request size is tracked + // incrementally. The boundaries are byte-identical to measuring + // bulkWriteRequestSize(current+record) on every append (see + // TestChunkOutboxRecordsMatchesLegacyBoundaries), without re-encoding the + // whole pending chunk per record — that was O(n^2) in bytes and dominated + // large --push-local-once seeds. chunks := make([][]outboxRecord, 0, 1) current := make([]outboxRecord, 0, len(records)) + var sizer bulkWriteRequestSizer for _, record := range records { - candidate := append(append([]outboxRecord(nil), current...), record) - if len(current) > 0 && bulkWriteRequestSize(outboxRecordsAsBulkFiles(candidate)) > maxBytes { + itemSize, ok := bulkWriteFileEncodedSize(outboxRecordAsBulkFile(record)) + if len(current) > 0 && sizer.sizeWith(itemSize, ok) > maxBytes { chunks = append(chunks, append([]outboxRecord(nil), current...)) current = current[:0] + sizer = bulkWriteRequestSizer{} } current = append(current, record) + sizer.add(itemSize, ok) } if len(current) > 0 { chunks = append(chunks, current) @@ -4570,21 +4721,25 @@ func chunkOutboxRecords(records []outboxRecord, maxBytes int64) [][]outboxRecord func outboxRecordsAsBulkFiles(records []outboxRecord) []BulkWriteFile { files := make([]BulkWriteFile, 0, len(records)) for _, record := range records { - files = append(files, BulkWriteFile{ - Path: record.RemotePath, - Type: normalizeRemoteType(record.Type), - Target: record.Target, - Mode: record.Mode, - ContentType: record.ContentType, - Content: record.Content, - Encoding: record.Encoding, - ContentIdentity: outboxRecordContentIdentity(record), - IfMatch: record.ExpectedRevision, - }) + files = append(files, outboxRecordAsBulkFile(record)) } return files } +func outboxRecordAsBulkFile(record outboxRecord) BulkWriteFile { + return BulkWriteFile{ + Path: record.RemotePath, + Type: normalizeRemoteType(record.Type), + Target: record.Target, + Mode: record.Mode, + ContentType: record.ContentType, + Content: record.Content, + Encoding: record.Encoding, + ContentIdentity: outboxRecordContentIdentity(record), + IfMatch: record.ExpectedRevision, + } +} + // outboxRecordContentIdentity derives the server-side dedupe key for a durable // outbox record. Writeback "create draft" paths must dedupe on // (workspace, path, content hash) — the same identity used by the in-flight @@ -4608,20 +4763,24 @@ func outboxRecordContentIdentity(record outboxRecord) *ContentIdentity { func bulkWriteFilesForPending(workspaceID string, pending []pendingBulkWrite) []BulkWriteFile { files := make([]BulkWriteFile, 0, len(pending)) for _, pendingWrite := range pending { - files = append(files, BulkWriteFile{ - Path: pendingWrite.remotePath, - Type: normalizeRemoteType(pendingWrite.snapshot.Type), - Target: pendingWrite.snapshot.Target, - Mode: pendingWrite.snapshot.Mode, - ContentType: pendingWrite.snapshot.ContentType, - Content: pendingWrite.snapshot.WireContent, - Encoding: pendingWrite.snapshot.Encoding, - ContentIdentity: mountWritebackCreateDraftContentIdentity(workspaceID, pendingWrite.remotePath, pendingWrite.snapshot.Hash), - }) + files = append(files, bulkWriteFileForPending(workspaceID, pendingWrite)) } return files } +func bulkWriteFileForPending(workspaceID string, pendingWrite pendingBulkWrite) BulkWriteFile { + return BulkWriteFile{ + Path: pendingWrite.remotePath, + Type: normalizeRemoteType(pendingWrite.snapshot.Type), + Target: pendingWrite.snapshot.Target, + Mode: pendingWrite.snapshot.Mode, + ContentType: pendingWrite.snapshot.ContentType, + Content: pendingWrite.snapshot.WireContent, + Encoding: pendingWrite.snapshot.Encoding, + ContentIdentity: mountWritebackCreateDraftContentIdentity(workspaceID, pendingWrite.remotePath, pendingWrite.snapshot.Hash), + } +} + func hasSymlinkBulkWrite(files []BulkWriteFile) bool { for _, file := range files { if isSymlinkType(file.Type) { @@ -4681,13 +4840,16 @@ func chunkPendingBulkWrites(workspaceID string, pending []pendingBulkWrite, maxB } chunks := make([][]pendingBulkWrite, 0, 1) current := make([]pendingBulkWrite, 0, len(pending)) + var sizer bulkWriteRequestSizer for _, item := range pending { - candidate := append(append([]pendingBulkWrite(nil), current...), item) - if len(current) > 0 && bulkWriteRequestSize(bulkWriteFilesForPending(workspaceID, candidate)) > maxBytes { + itemSize, ok := bulkWriteFileEncodedSize(bulkWriteFileForPending(workspaceID, item)) + if len(current) > 0 && sizer.sizeWith(itemSize, ok) > maxBytes { chunks = append(chunks, append([]pendingBulkWrite(nil), current...)) current = current[:0] + sizer = bulkWriteRequestSizer{} } current = append(current, item) + sizer.add(itemSize, ok) } if len(current) > 0 { chunks = append(chunks, current) @@ -4706,6 +4868,55 @@ func bulkWriteRequestSize(files []BulkWriteFile) int64 { return int64(len(data)) } +// bulkWriteRequestEnvelopeBytes is len(`{"files":[`) + len(`]}`): the fixed +// framing json.Marshal emits around a non-empty files array. +const bulkWriteRequestEnvelopeBytes int64 = int64(len(`{"files":[`) + len(`]}`)) + +// bulkWriteFileEncodedSize returns the length of file's JSON encoding as an +// element of the bulk request array. encoding/json encodes slice elements +// with the same encoder state (including HTML escaping) as a standalone +// json.Marshal of the element, so the request body is exactly the envelope +// plus each element plus one comma between elements. +func bulkWriteFileEncodedSize(file BulkWriteFile) (int64, bool) { + data, err := json.Marshal(file) + if err != nil { + return 0, false + } + return int64(len(data)), true +} + +// bulkWriteRequestSizer incrementally tracks bulkWriteRequestSize for a +// growing files slice. It reproduces bulkWriteRequestSize exactly, including +// its "0 on encode error" result, so callers keep identical batch boundaries. +type bulkWriteRequestSizer struct { + count int + itemBytes int64 + invalid bool +} + +func (z *bulkWriteRequestSizer) add(itemSize int64, ok bool) { + if !ok { + z.invalid = true + } + z.count++ + z.itemBytes += itemSize +} + +// size is bulkWriteRequestSize of the files added so far (non-empty). +func (z bulkWriteRequestSizer) size() int64 { + if z.invalid || z.count == 0 { + return 0 + } + return bulkWriteRequestEnvelopeBytes + z.itemBytes + int64(z.count-1) +} + +// sizeWith is bulkWriteRequestSize of the files added so far plus one more +// element of itemSize bytes. +func (z bulkWriteRequestSizer) sizeWith(itemSize int64, ok bool) int64 { + z.add(itemSize, ok) + return z.size() +} + func (s *Syncer) reconcileBulkWrite(ctx context.Context, pendingWrite pendingBulkWrite, revision string) error { tracked := pendingWrite.tracked contentType := strings.TrimSpace(pendingWrite.snapshot.ContentType) @@ -5169,6 +5380,43 @@ func bulkWriteErrorAsError(writeErr BulkWriteError) error { return fmt.Errorf("bulk write failed for %s", normalizeRemotePath(writeErr.Path)) } +// defaultBulkUploadConcurrency bounds concurrent /fs/bulk POSTs while one +// flush drains several outbox chunks. Two full 8 MiB chunks fit the Cloud +// workspace's 16 MiB in-flight body budget; a third would only trade time for +// 429 workspace_busy retries. +const defaultBulkUploadConcurrency = 2 + +// maxBulkUploadConcurrency caps RELAYFILE_MOUNT_BULK_UPLOAD_CONCURRENCY. +const maxBulkUploadConcurrency = 8 + +func (s *Syncer) bulkUploadConcurrencyValue() int { + if s.bulkUploadConcurrency > 0 { + return s.bulkUploadConcurrency + } + return defaultBulkUploadConcurrency +} + +// bulkUploadConcurrencyFromEnv reads RELAYFILE_MOUNT_BULK_UPLOAD_CONCURRENCY: +// a positive integer (1 restores strictly serial uploads), clamped to +// maxBulkUploadConcurrency. Invalid values keep the default. +func bulkUploadConcurrencyFromEnv(logger Logger) int { + raw := strings.TrimSpace(os.Getenv("RELAYFILE_MOUNT_BULK_UPLOAD_CONCURRENCY")) + if raw == "" { + return defaultBulkUploadConcurrency + } + v, err := strconv.Atoi(raw) + if err != nil || v <= 0 { + if logger != nil { + logger.Printf("ignoring invalid RELAYFILE_MOUNT_BULK_UPLOAD_CONCURRENCY=%q", raw) + } + return defaultBulkUploadConcurrency + } + if v > maxBulkUploadConcurrency { + return maxBulkUploadConcurrency + } + return v +} + func (s *Syncer) bulkFlushThresholdValue() int { if s.bulkFlushThreshold > 0 { return s.bulkFlushThreshold @@ -10759,10 +11007,13 @@ func (s *Syncer) pushLocal(ctx context.Context) (map[string]struct{}, error) { if !s.recoverStartupDrift && s.localWatcherActive && !s.pollLocalChanges { return conflicted, nil } - localFiles, err := s.scanLocalFiles() + localFiles, capturedContent, err := s.scanLocalFilesCapturing(true) if err != nil { return nil, err } + if s.retainPushLocalScan { + s.lastPushLocalScan = localFiles + } localRemotePaths := make([]string, 0, len(localFiles)) for remotePath := range localFiles { @@ -10896,11 +11147,19 @@ func (s *Syncer) pushLocal(ctx context.Context) (map[string]struct{}, error) { s.state.Files[remotePath] = tracked continue } - fullSnapshot, err := s.readLocalSnapshot(localPath, true) - if err != nil { - return nil, err + if fullSnapshot, ok := capturedContent[remotePath]; ok && !exists { + // The scan already read this untracked file whole; its hash in + // localFiles came from that same read. Release the map's reference + // so content is retained only as long as the pending batch needs it. + delete(capturedContent, remotePath) + snapshot = fullSnapshot + } else { + fullSnapshot, err := s.readLocalSnapshot(localPath, true) + if err != nil { + return nil, err + } + snapshot = fullSnapshot } - snapshot = fullSnapshot pendingWrite, err := s.preparePendingBulkWrite(ctx, remotePath, localPath, snapshot, tracked, exists) if err != nil { return nil, err @@ -11022,11 +11281,40 @@ func (s *Syncer) applyWriteDenied(ctx context.Context, remotePath, localPath str } func (s *Syncer) scanLocalFiles() (map[string]localSnapshot, error) { + results, _, err := s.scanLocalFilesCapturing(false) + return results, err +} + +const ( + // localScanCaptureMaxFileBytes bounds which untracked files pushLocal's + // scan keeps in memory instead of hashing and then re-reading them. + localScanCaptureMaxFileBytes int64 = 1 << 20 + // localScanCaptureBudgetBytes bounds the total raw bytes captured by one + // scan. Beyond it the scan falls back to hash-only reads, so a huge + // untracked tree costs no more memory than before. + localScanCaptureBudgetBytes int64 = 64 << 20 +) + +// scanLocalFilesCapturing is scanLocalFiles that, when captureUntracked is +// set, reads small untracked regular files whole during the walk and returns +// those full snapshots in captured (keyed by remote path). pushLocal must +// read every untracked file in full anyway; capturing it here turns the +// hash-then-reread double read into one read. The scan map itself keeps its +// usual hash-only shape, derived from the very same read, so the hash and the +// captured content always agree. Files that change after the read are +// indistinguishable from files that change right after a fresh re-read and +// are picked up by the next scan or watcher event. +func (s *Syncer) scanLocalFilesCapturing(captureUntracked bool) (map[string]localSnapshot, map[string]localSnapshot, error) { results := map[string]localSnapshot{} + var captured map[string]localSnapshot + captureBudget := localScanCaptureBudgetBytes + if captureUntracked && !s.lowMemory { + captured = map[string]localSnapshot{} + } githubPathIndex := s.githubWorkingTreePathIndex() statePathAbs, err := filepath.Abs(s.stateFile) if err != nil { - return nil, err + return nil, nil, err } err = filepath.WalkDir(s.localRoot, func(path string, d os.DirEntry, walkErr error) error { if walkErr != nil { @@ -11139,9 +11427,15 @@ func (s *Syncer) scanLocalFiles() (map[string]localSnapshot, error) { s.state.Counters.SkippedOversizeWriteback++ return nil } + skipLazyUntracked := s.shouldSkipLazyUntrackedPush(remotePath) + capture := false + if captured != nil && !skipLazyUntracked && info.Size() <= localScanCaptureMaxFileBytes && info.Size() <= captureBudget { + _, tracked := s.state.Files[remotePath] + capture = !tracked + } var snapshot localSnapshot s.runReservedSyncIO(func() { - snapshot, err = s.readLocalSnapshot(path, false) + snapshot, err = s.readLocalSnapshot(path, capture) }) if err != nil { if errors.Is(err, os.ErrNotExist) { @@ -11149,7 +11443,18 @@ func (s *Syncer) scanLocalFiles() (map[string]localSnapshot, error) { } return err } - if s.shouldSkipLazyUntrackedPush(remotePath) { + if capture && !isSymlinkType(snapshot.Type) { + captureBudget -= int64(len(snapshot.RawContent)) + captured[remotePath] = snapshot + snapshot = localSnapshot{ + ContentType: snapshot.ContentType, + Type: snapshot.Type, + Mode: snapshot.Mode, + Hash: snapshot.Hash, + LocalPath: snapshot.LocalPath, + } + } + if skipLazyUntracked { // Only flag the snapshot here; do NOT increment // Counters.SkippedLazyUntrackedPush in this function. // scanLocalFiles also runs from savePublicState (status-display @@ -11165,9 +11470,9 @@ func (s *Syncer) scanLocalFiles() (map[string]localSnapshot, error) { return nil }) if err != nil { - return nil, err + return nil, nil, err } - return results, nil + return results, captured, nil } // githubAdapterCreateCommandPathPatterns enumerate the GitHub Relayfile @@ -11600,6 +11905,19 @@ func (s *Syncer) savePublicStateWithLocalScan(scanLocal bool) error { return err } } + return s.savePublicStateWithLocalFiles(currentFiles, scanLocal) +} + +// saveStateWithLocalFiles is saveState with a caller-supplied local scan +// (the hash-only map scanLocalFiles returns) instead of a fresh tree walk. +func (s *Syncer) saveStateWithLocalFiles(currentFiles map[string]localSnapshot) error { + if err := s.savePrivateState(); err != nil { + return err + } + return s.savePublicStateWithLocalFiles(currentFiles, !s.lowMemory) +} + +func (s *Syncer) savePublicStateWithLocalFiles(currentFiles map[string]localSnapshot, scanLocal bool) error { conflictsByPath, pendingConflicts, err := s.listConflictArtifacts() if err != nil { return err diff --git a/internal/mountsync/syncer_test.go b/internal/mountsync/syncer_test.go index e07ede4b..01b8a49f 100644 --- a/internal/mountsync/syncer_test.go +++ b/internal/mountsync/syncer_test.go @@ -12326,7 +12326,7 @@ var ( mountsyncTestPrivateKey *rsa.PrivateKey ) -func ensureMountsyncJWTVerifier(t *testing.T) { +func ensureMountsyncJWTVerifier(t testing.TB) { t.Helper() mountsyncTestJWKSOnce.Do(func() { @@ -12355,7 +12355,7 @@ func ensureMountsyncJWTVerifier(t *testing.T) { }) } -func newMountsyncAPIHandler(t *testing.T, store *relayfile.Store) http.Handler { +func newMountsyncAPIHandler(t testing.TB, store *relayfile.Store) http.Handler { t.Helper() ensureMountsyncJWTVerifier(t) @@ -12369,7 +12369,7 @@ func newMountsyncAPIHandler(t *testing.T, store *relayfile.Store) http.Handler { return handler } -func mustMountsyncTestJWT(t *testing.T, secret, workspaceID, agentName string, scopes []string, exp time.Time) string { +func mustMountsyncTestJWT(t testing.TB, secret, workspaceID, agentName string, scopes []string, exp time.Time) string { t.Helper() _ = secret ensureMountsyncJWTVerifier(t) From 8d123436af28ca0db748c8e1801ee4bccea4555a Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sat, 10 Oct 2026 16:33:51 -0700 Subject: [PATCH 2/2] fix(mountsync): address review on push perf (stale capture, stop-on-failure, removed paths) - Captured untracked-file content is used only if the file is provably unchanged since the scan read it (same inode, size, mtime, mode as the pre-read Lstat); otherwise it is re-read so the latest bytes upload. - flushOutboxChunks inspects every completed upload, not only the head, before refilling: a transport error or a failed receipt stops dispatch for good, and a completed chunk with per-path errors pauses dispatch until it is applied. No chunk is sent once a failure is observable. - PushLocalAndFlushOnce drops paths whose files the flush removed (e.g. a schema-rejected new draft) from the reused scan before publishing public state; only untracked/DeletePending entries are Lstat'ed. Co-Authored-By: Claude Opus 5.5 --- .../mountsync/bulk_upload_concurrency_test.go | 120 ++++++++++++-- .../mountsync/push_local_read_once_test.go | 102 ++++++++++++ internal/mountsync/syncer.go | 153 ++++++++++++++---- 3 files changed, 338 insertions(+), 37 deletions(-) diff --git a/internal/mountsync/bulk_upload_concurrency_test.go b/internal/mountsync/bulk_upload_concurrency_test.go index 0caad7aa..92b57dc8 100644 --- a/internal/mountsync/bulk_upload_concurrency_test.go +++ b/internal/mountsync/bulk_upload_concurrency_test.go @@ -2,6 +2,7 @@ package mountsync import ( "context" + "errors" "fmt" "net/http" "path/filepath" @@ -30,6 +31,16 @@ type concurrentBulkClient struct { failFirstPath string failErr error failed bool + // holdOthers delays every non-failing POST, so a failing one completes + // first and its result is buffered while an earlier head is in flight. + holdOthers time.Duration + // failureSeen is set once a failing POST has returned; every POST of + // the initial window has started by then (the failure is delayed by + // failDelay), so startsAfterFailure counts only refills sent after the + // failure was observable. + failDelay time.Duration + failureSeen bool + startsAfterFailure int } func newConcurrentBulkClient(overlap int) *concurrentBulkClient { @@ -44,6 +55,9 @@ func newConcurrentBulkClient(overlap int) *concurrentBulkClient { func (c *concurrentBulkClient) WriteFilesBulk(ctx context.Context, workspaceID string, files []BulkWriteFile) (BulkWriteResponse, error) { c.mu.Lock() + if c.failureSeen { + c.startsAfterFailure++ + } c.calls++ c.inFlight++ if c.inFlight > c.maxInFlight { @@ -68,10 +82,20 @@ func (c *concurrentBulkClient) WriteFilesBulk(ctx context.Context, workspaceID s c.mu.Unlock() // Hold briefly so concurrently dispatched chunks really overlap. - time.Sleep(5 * time.Millisecond) + switch { + case fail: + time.Sleep(c.failDelay) + case c.holdOthers > 0: + time.Sleep(c.holdOthers) + default: + time.Sleep(5 * time.Millisecond) + } c.mu.Lock() c.inFlight-- + if fail { + c.failureSeen = true + } c.mu.Unlock() if fail { return BulkWriteResponse{}, c.failErr @@ -192,8 +216,8 @@ func TestFlushOutboxChunksSerialWhenConcurrencyIsOne(t *testing.T) { } // A 429 workspace_busy that outlives the client's own Retry-After retries -// fails only its chunk: those records stay pending with backoff, chunks -// already in flight are still acked (never re-sent), no later chunk is +// fails only its chunk: those records stay pending with backoff, the chunk +// in flight beside it is still acked (never re-sent), no later chunk is // dispatched, and the next flush delivers the rest exactly once. func TestFlushOutboxChunksWorkspaceBusyKeepsAccountingExact(t *testing.T) { t.Setenv("RELAYFILE_MAX_WRITEBACK_BATCH_BYTES", "4096") @@ -233,11 +257,12 @@ func TestFlushOutboxChunksWorkspaceBusyKeepsAccountingExact(t *testing.T) { t.Fatalf("%s failed-chunk record = attempts %d next %q err %q, want one backed-off attempt", path, record.AttemptCount, record.NextAttemptAt, record.LastError) } } - // With two in flight, the third chunk was dispatched alongside the failing - // second one and the server accepted it: it must be acked, not resent. + // The failure was observed before the window refilled, so the third + // chunk was never sent: it stays pending with no attempt recorded. for _, path := range paths[6:9] { - if _, ok := acked[path]; !ok { - t.Fatalf("%s was accepted by the server while the failed chunk was in flight but not acked", path) + record, ok := pending[path] + if !ok || record.AttemptCount != 0 || client.uploads[path] != 0 { + t.Fatalf("%s after the failed chunk: pending=%v attempts=%d uploads=%d, want never sent", path, ok, record.AttemptCount, client.uploads[path]) } } for path := range pending { @@ -245,8 +270,8 @@ func TestFlushOutboxChunksWorkspaceBusyKeepsAccountingExact(t *testing.T) { t.Fatalf("pending %s was already accepted by the server %d time(s)", path, client.uploads[path]) } } - if client.calls >= 10 { - t.Fatalf("chunks after the failure were still dispatched: %d POSTs", client.calls) + if client.calls != 2 { + t.Fatalf("chunks after the failure were dispatched: %d POSTs, want 2", client.calls) } // The next drain (FlushOutboxOnce forces due) delivers everything left. @@ -338,3 +363,80 @@ func (c *pathOrderClient) WriteFilesBulk(ctx context.Context, workspaceID string c.mu.Unlock() return response, err } + +// With concurrency above 1, once any chunk has failed — a workspace_busy or +// transport error that completed while an earlier chunk was still in flight, +// or a failed receipt in a chunk being prepared — no further chunk may be +// sent, matching the serial loop's stop-on-first-error. +func TestFlushOutboxChunksSendsNothingAfterFirstFailure(t *testing.T) { + failures := map[string]func(c *concurrentBulkClient, s *Syncer, paths []string){ + "workspace_busy": func(c *concurrentBulkClient, _ *Syncer, paths []string) { + c.failFirstPath = paths[3] + c.failErr = &HTTPError{StatusCode: http.StatusTooManyRequests, Code: "workspace_busy", Message: "workspace busy"} + }, + "transport": func(c *concurrentBulkClient, _ *Syncer, paths []string) { + c.failFirstPath = paths[3] + c.failErr = errors.New("connection reset by peer") + }, + "receipt": func(c *concurrentBulkClient, s *Syncer, _ []string) { + // The second chunk holds a record whose provider dispatch already + // failed plus two new records; only those two may still upload. + record, err := s.readOutboxRecord(s.pendingOutboxPath("mountcmd_conc_003")) + if err != nil { + t.Fatalf("read seeded record: %v", err) + } + record.OpID = "op_failed" + if err := s.saveOutboxRecord(record); err != nil { + t.Fatalf("save receipt record: %v", err) + } + c.fakeClient.operations = map[string]OperationStatus{"op_failed": {OpID: "op_failed", Status: "failed"}} + }, + } + for name, inject := range failures { + for _, limit := range []int{2, 8} { + t.Run(fmt.Sprintf("%s/concurrency=%d", name, limit), func(t *testing.T) { + t.Setenv("RELAYFILE_MAX_WRITEBACK_BATCH_BYTES", "4096") + client := newConcurrentBulkClient(1) + client.holdOthers = 80 * time.Millisecond + client.failDelay = 10 * time.Millisecond + syncer, localDir := newConcurrencySyncer(t, client, limit) + paths := seedConcurrencyOutbox(t, syncer, 30) + inject(client, syncer, paths) + + if err := syncer.FlushOutboxOnce(context.Background()); err == nil { + t.Fatal("FlushOutboxOnce succeeded despite a failed chunk") + } + if client.startsAfterFailure != 0 { + t.Fatalf("%d chunk(s) were sent after the failure was observable (calls=%d)", client.startsAfterFailure, client.calls) + } + if name == "receipt" && client.calls != 2 { + // The receipt fails while preparing the second chunk: the + // first chunk and the second chunk's own new records are + // the only POSTs, whatever the concurrency. + t.Fatalf("receipt failure: %d POSTs, want 2", client.calls) + } + if limit == 2 && client.calls != 2 { + t.Fatalf("concurrency 2: %d POSTs, want only the two dispatched before the failure", client.calls) + } + if client.calls > limit { + t.Fatalf("%d POSTs exceed the initial window of %d", client.calls, limit) + } + acked := outboxRecordsByPathForTest(t, localDir, "acked") + pending := outboxRecordsByPathForTest(t, localDir, "pending") + for path := range pending { + if _, both := acked[path]; both { + t.Fatalf("%s both acked and pending", path) + } + if client.uploads[path] != 0 { + t.Fatalf("pending %s was accepted by the server", path) + } + } + for path := range acked { + if client.uploads[path] != 1 { + t.Fatalf("acked %s accepted %d times", path, client.uploads[path]) + } + } + }) + } + } +} diff --git a/internal/mountsync/push_local_read_once_test.go b/internal/mountsync/push_local_read_once_test.go index 743ef1f5..608606f2 100644 --- a/internal/mountsync/push_local_read_once_test.go +++ b/internal/mountsync/push_local_read_once_test.go @@ -142,3 +142,105 @@ func TestPushLocalLowMemoryDoesNotCaptureContent(t *testing.T) { t.Fatalf("low-memory scan captured %d files", len(captured)) } } + +// A file captured by the scan and edited before the push loop reaches it +// must upload its latest bytes: the capture is revalidated (inode, size, +// mtime) and a changed file is re-read. The one-shot drain does not rescan, +// so stale captured bytes would otherwise be the final remote content. +func TestPushLocalAndFlushOnceUploadsFileEditedAfterScan(t *testing.T) { + localDir := t.TempDir() + aPath := filepath.Join(localDir, "a.txt") + bPath := filepath.Join(localDir, "b.txt") + if err := os.WriteFile(aPath, []byte("v1\n"), 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(bPath, []byte("b\n"), 0o644); err != nil { + t.Fatal(err) + } + client := &fakeClient{files: map[string]RemoteFile{}} + syncer, err := NewSyncer(client, SyncerOptions{ + WorkspaceID: "ws_capture_stale", + RemoteRoot: "/", + LocalRoot: localDir, + }) + if err != nil { + t.Fatalf("NewSyncer: %v", err) + } + edited := false + syncer.readLocalSnapshotFn = func(path string, includeContent bool) (localSnapshot, error) { + snapshot, err := readLocalSnapshotLimitedUnderRoot(localDir, path, includeContent, maxWritebackBytes()) + // The walk is lexical: a.txt is captured first, then b.txt. Edit + // a.txt while b.txt is being scanned, i.e. after a.txt's capture and + // before the push loop uses it. + if path == bPath && !edited { + edited = true + if err := os.WriteFile(aPath, []byte("v2 edited after scan\n"), 0o644); err != nil { + t.Fatal(err) + } + } + return snapshot, err + } + + if err := syncer.PushLocalAndFlushOnce(context.Background()); err != nil { + t.Fatalf("PushLocalAndFlushOnce: %v", err) + } + if !edited { + t.Fatal("test hook never edited a.txt") + } + got, ok := client.files["/a.txt"] + if !ok || got.Content != "v2 edited after scan\n" { + t.Fatalf("remote /a.txt = %q (present=%v), want the post-scan edit", got.Content, ok) + } + if hash := syncer.state.Files["/a.txt"].Hash; hash != hashBytes([]byte("v2 edited after scan\n")) { + t.Fatalf("tracked hash %s is not the uploaded latest content", hash) + } +} + +// When the flush removes a local file — a new draft rejected with +// schema_validation_failed that has no remote version to restore — the +// public state derived from the reused pre-flush scan must not list it. +func TestPushLocalAndFlushOnceOmitsFilesRemovedByFlushFromPublicState(t *testing.T) { + const draft = "/github/repos/acme/api/pulls/42/reviews/draft.json" + client := &fakeClient{ + files: map[string]RemoteFile{}, + bulkWriteResponseFunc: func(ctx context.Context, workspaceID string, files []BulkWriteFile) (BulkWriteResponse, error) { + return BulkWriteResponse{ErrorCount: 1, Errors: []BulkWriteError{{ + Path: draft, Code: "schema_validation_failed", Message: "body.event is required", + }}}, nil + }, + } + localDir := t.TempDir() + syncer, err := NewSyncer(client, SyncerOptions{ + WorkspaceID: "ws_schema_removed_public_state", + RemoteRoot: "/github", + LocalRoot: localDir, + }) + if err != nil { + t.Fatalf("NewSyncer: %v", err) + } + localPath := filepath.Join(localDir, "repos", "acme", "api", "pulls", "42", "reviews", "draft.json") + if err := os.MkdirAll(filepath.Dir(localPath), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(localPath, []byte(`{"body":"no event"}`), 0o644); err != nil { + t.Fatal(err) + } + keep := filepath.Join(localDir, "notes.txt") + if err := os.WriteFile(keep, []byte("kept\n"), 0o644); err != nil { + t.Fatal(err) + } + + if err := syncer.PushLocalAndFlushOnce(context.Background()); err != nil { + t.Fatalf("PushLocalAndFlushOnce: %v", err) + } + if _, err := os.Lstat(localPath); !os.IsNotExist(err) { + t.Fatalf("schema-rejected new draft should have been removed locally, stat err=%v", err) + } + state := readPublicState(t, localDir) + if entry, listed := state.Files[draft]; listed { + t.Fatalf("removed draft still published as %q", entry.Status) + } + if state.PendingWriteback != 0 { + t.Fatalf("pendingWriteback = %d, want 0 after the drain", state.PendingWriteback) + } +} diff --git a/internal/mountsync/syncer.go b/internal/mountsync/syncer.go index 3a91d856..f8f1380b 100644 --- a/internal/mountsync/syncer.go +++ b/internal/mountsync/syncer.go @@ -2992,12 +2992,12 @@ func (s *Syncer) PushLocalAndFlushOnce(ctx context.Context) error { } s.markSyncSuccess() if s.lastPushLocalScan != nil { - // pushLocal just walked and hashed the tree. The flush that followed - // only uploads; its conflict/read-only handling never changes which - // paths exist locally or whether they are writeback-skipped — the - // only facts the public state reads from a scan. Reuse that scan - // rather than hashing every file a third time. - return s.saveStateWithLocalFiles(s.lastPushLocalScan) + // pushLocal just walked and hashed the tree. Reuse that scan rather + // than hashing every file a third time. The public state reads only + // presence and skip status from it, and the flush can remove local + // files (e.g. materializeSchemaInvalid deletes a rejected new draft), + // so first drop entries whose files have since vanished. + return s.saveStateWithLocalFiles(s.dropVanishedPushScanEntries(s.lastPushLocalScan)) } return s.saveState() } @@ -4191,9 +4191,12 @@ func (s *Syncer) flushOutboxRecords(ctx context.Context, conflicted map[string]s // serial loop, and a record is acked only after the server confirmed it; // - a chunk sharing any path with an in-flight chunk waits for it, so two // commands for one path are never in flight together; -// - after the first error no further chunk is dispatched (as before), but -// uploads already in flight are still applied, so a write the server -// accepted is acked instead of being re-sent. +// - completed uploads (not only the head) are inspected before every +// refill: a transport error or a failed receipt stops dispatch for good, +// and a completed chunk with per-path errors pauses dispatch until it has +// been applied, so no chunk is sent once a failure is observable (the +// serial loop's stop-on-first-error). Uploads already in flight are +// still applied, so a write the server accepted is acked, not re-sent. func (s *Syncer) flushOutboxChunks(ctx context.Context, chunks [][]outboxRecord, conflicted map[string]struct{}, settleReceipts, unlockDuringUpload bool) error { limit := s.bulkUploadConcurrencyValue() if len(chunks) <= 1 || limit <= 1 || unlockDuringUpload { @@ -4214,6 +4217,7 @@ func (s *Syncer) flushOutboxChunks(ctx context.Context, chunks [][]outboxRecord, paths []string started time.Time done chan uploadResult + result *uploadResult // set once the upload has completed } var ( queue []*inflightChunk @@ -4230,18 +4234,54 @@ func (s *Syncer) flushOutboxChunks(ctx context.Context, chunks [][]outboxRecord, } return false } + // observeCompleted collects every upload that has already finished — + // not only the head — so a failure is acted on before any refill. It + // returns true while dispatch must pause: a completed chunk carries + // per-path errors whose handling (applied in order) may still fail the + // flush. A transport error always fails its chunk, so it stops dispatch + // for good, exactly as the serial loop stops at the first failed chunk. + observeCompleted := func() (hold bool) { + for _, item := range queue { + if item.result == nil { + select { + case result := <-item.done: + item.result = &result + default: + continue + } + } + if item.result.err != nil { + stopDispatch = true + } + if len(item.result.response.Errors) > 0 || item.result.response.ErrorCount > 0 { + hold = true + } + } + return hold + } for { - for !stopDispatch && next < len(chunks) && len(queue) < limit && !conflictsInflight(chunks[next]) { + for next < len(chunks) && len(queue) < limit && !conflictsInflight(chunks[next]) { + if hold := observeCompleted(); hold || stopDispatch { + break + } chunk := chunks[next] next++ upload, done, err := s.prepareOutboxChunkUpload(ctx, chunk, settleReceipts) if done { if err != nil { - firstErr = err + if firstErr == nil { + firstErr = err + } stopDispatch = true } continue } + if upload.firstErr != nil { + // A receipt in this chunk already failed; the chunk's own new + // records still upload (as in the serial loop), but nothing + // after it may be sent. + stopDispatch = true + } item := &inflightChunk{upload: upload, started: time.Now(), done: make(chan uploadResult, 1)} for _, record := range upload.records { path := normalizeRemotePath(record.RemotePath) @@ -4259,7 +4299,11 @@ func (s *Syncer) flushOutboxChunks(ctx context.Context, chunks [][]outboxRecord, } head := queue[0] queue = queue[1:] - result := <-head.done + if head.result == nil { + result := <-head.done + head.result = &result + } + result := *head.result for _, path := range head.paths { if inflightBy[path]--; inflightBy[path] <= 0 { delete(inflightBy, path) @@ -11147,13 +11191,17 @@ func (s *Syncer) pushLocal(ctx context.Context) (map[string]struct{}, error) { s.state.Files[remotePath] = tracked continue } - if fullSnapshot, ok := capturedContent[remotePath]; ok && !exists { - // The scan already read this untracked file whole; its hash in - // localFiles came from that same read. Release the map's reference - // so content is retained only as long as the pending batch needs it. + if captured, ok := capturedContent[remotePath]; ok && !exists && captured.fresh(localPath) { + // The scan already read this untracked file whole and it is + // provably unchanged since (same inode, size and mtime as before + // that read). Release the map's reference so content is retained + // only as long as the pending batch needs it. delete(capturedContent, remotePath) - snapshot = fullSnapshot + snapshot = captured.snapshot } else { + // Not captured, or the file changed after the scan read it: read + // the current bytes, exactly as before content capture existed. + delete(capturedContent, remotePath) fullSnapshot, err := s.readLocalSnapshot(localPath, true) if err != nil { return nil, err @@ -11285,6 +11333,31 @@ func (s *Syncer) scanLocalFiles() (map[string]localSnapshot, error) { return results, err } +// capturedLocalFile is a full snapshot read during pushLocal's scan plus the +// Lstat taken immediately before that read. +type capturedLocalFile struct { + snapshot localSnapshot + info os.FileInfo +} + +// fresh reports whether localPath is still the same regular file, with the +// same size and modification time, that was observed before the captured +// read. Any write that began during or after the read moves mtime past the +// pre-read stat, so a match proves the captured bytes are still current. +func (c capturedLocalFile) fresh(localPath string) bool { + if c.info == nil { + return false + } + current, err := os.Lstat(localPath) + if err != nil || !current.Mode().IsRegular() { + return false + } + return os.SameFile(c.info, current) && + current.Size() == c.info.Size() && + current.ModTime().Equal(c.info.ModTime()) && + current.Mode() == c.info.Mode() +} + const ( // localScanCaptureMaxFileBytes bounds which untracked files pushLocal's // scan keeps in memory instead of hashing and then re-reading them. @@ -11297,19 +11370,19 @@ const ( // scanLocalFilesCapturing is scanLocalFiles that, when captureUntracked is // set, reads small untracked regular files whole during the walk and returns -// those full snapshots in captured (keyed by remote path). pushLocal must -// read every untracked file in full anyway; capturing it here turns the -// hash-then-reread double read into one read. The scan map itself keeps its -// usual hash-only shape, derived from the very same read, so the hash and the -// captured content always agree. Files that change after the read are -// indistinguishable from files that change right after a fresh re-read and -// are picked up by the next scan or watcher event. -func (s *Syncer) scanLocalFilesCapturing(captureUntracked bool) (map[string]localSnapshot, map[string]localSnapshot, error) { +// those full snapshots in captured (keyed by remote path), together with the +// file's pre-read Lstat. pushLocal must read every untracked file in full +// anyway; capturing it here turns the hash-then-reread double read into one +// read. The scan map keeps its usual hash-only shape, derived from the very +// same read, so the hash and the captured content always agree. Consumers +// must check capturedLocalFile.fresh before using the content; a file that +// changed after the scan read it is re-read, so the latest bytes are pushed. +func (s *Syncer) scanLocalFilesCapturing(captureUntracked bool) (map[string]localSnapshot, map[string]capturedLocalFile, error) { results := map[string]localSnapshot{} - var captured map[string]localSnapshot + var captured map[string]capturedLocalFile captureBudget := localScanCaptureBudgetBytes if captureUntracked && !s.lowMemory { - captured = map[string]localSnapshot{} + captured = map[string]capturedLocalFile{} } githubPathIndex := s.githubWorkingTreePathIndex() statePathAbs, err := filepath.Abs(s.stateFile) @@ -11445,7 +11518,7 @@ func (s *Syncer) scanLocalFilesCapturing(captureUntracked bool) (map[string]loca } if capture && !isSymlinkType(snapshot.Type) { captureBudget -= int64(len(snapshot.RawContent)) - captured[remotePath] = snapshot + captured[remotePath] = capturedLocalFile{snapshot: snapshot, info: info} snapshot = localSnapshot{ ContentType: snapshot.ContentType, Type: snapshot.Type, @@ -11908,6 +11981,30 @@ func (s *Syncer) savePublicStateWithLocalScan(scanLocal bool) error { return s.savePublicStateWithLocalFiles(currentFiles, scanLocal) } +// dropVanishedPushScanEntries removes, from a scan taken before a flush, the +// paths whose local files no longer exist. Only entries the public state +// would publish from the scan itself are checked — untracked paths (listed as +// writeback-pending/skipped) and tracked DeletePending paths — so after a +// successful drain this is a handful of Lstats, not a tree walk. +func (s *Syncer) dropVanishedPushScanEntries(scan map[string]localSnapshot) map[string]localSnapshot { + for remotePath, snapshot := range scan { + if tracked, ok := s.state.Files[remotePath]; ok && !tracked.DeletePending { + continue + } + localPath := snapshot.LocalPath + if strings.TrimSpace(localPath) == "" { + var err error + if localPath, err = s.remoteToLocalPath(remotePath); err != nil { + continue + } + } + if _, err := os.Lstat(localPath); errors.Is(err, os.ErrNotExist) { + delete(scan, remotePath) + } + } + return scan +} + // saveStateWithLocalFiles is saveState with a caller-supplied local scan // (the hash-only map scanLocalFiles returns) instead of a fresh tree walk. func (s *Syncer) saveStateWithLocalFiles(currentFiles map[string]localSnapshot) error {