From 6290ab3995119114c3c27472a1194bde22fbb17d Mon Sep 17 00:00:00 2001 From: Khaliq Date: Mon, 24 Aug 2026 11:36:41 +0200 Subject: [PATCH] fix: chunk bulk writes below SQL variable limit --- .../completed/2026-08/traj_hhthwa7xxikz.json | 78 +++++++++++++++++++ .../completed/2026-08/traj_hhthwa7xxikz.md | 33 ++++++++ .trajectories/index.json | 9 ++- internal/mountsync/http_client_test.go | 71 +++++++++++++++++ internal/mountsync/syncer.go | 36 +++++++-- internal/mountsync/syncer_test.go | 76 ++++++++++++++++-- packages/sdk/python/src/relayfile/client.py | 68 +++++++++++++--- packages/sdk/python/tests/test_client.py | 70 +++++++++++++++++ packages/sdk/typescript/src/client.test.ts | 30 +++++++ packages/sdk/typescript/src/client.ts | 45 ++++++++--- 10 files changed, 479 insertions(+), 37 deletions(-) create mode 100644 .trajectories/completed/2026-08/traj_hhthwa7xxikz.json create mode 100644 .trajectories/completed/2026-08/traj_hhthwa7xxikz.md diff --git a/.trajectories/completed/2026-08/traj_hhthwa7xxikz.json b/.trajectories/completed/2026-08/traj_hhthwa7xxikz.json new file mode 100644 index 00000000..29828950 --- /dev/null +++ b/.trajectories/completed/2026-08/traj_hhthwa7xxikz.json @@ -0,0 +1,78 @@ +{ + "id": "traj_hhthwa7xxikz", + "version": 1, + "task": { + "title": "Fix #445 SQL variable limit data loss", + "source": { + "system": "plain", + "id": "AgentWorkforce/relayfile#445" + } + }, + "status": "completed", + "startedAt": "2026-08-24T09:12:49.140Z", + "completedAt": "2026-08-24T09:35:03.059Z", + "agents": [ + { + "name": "default", + "role": "lead", + "joinedAt": "2026-08-24T09:28:38.146Z" + } + ], + "chapters": [ + { + "id": "chap_iocnaxzx9nvs", + "title": "Work", + "agentName": "default", + "startedAt": "2026-08-24T09:28:38.146Z", + "endedAt": "2026-08-24T09:35:03.059Z", + "events": [ + { + "ts": 1787563718150, + "type": "decision", + "content": "Cap bulk writes at 200 files across durable outbox and client boundaries: Cap bulk writes at 200 files across durable outbox and client boundaries", + "raw": { + "question": "Cap bulk writes at 200 files across durable outbox and client boundaries", + "chosen": "Cap bulk writes at 200 files across durable outbox and client boundaries", + "alternatives": [], + "reasoning": "The outbox maintenance path chunks only by 8 MiB, so 1,454 small records can be reassembled into one request. A fixed 200-file cap stays well below SQLite's conservative 999-variable limit, preserves byte chunking, and protects foreground, maintenance, and direct client callers." + }, + "significance": "high" + }, + { + "ts": 1787564102654, + "type": "reflection", + "content": "Reproduced the production signature with 1,001 small workspace files; count chunking is green across Go mount, direct Go HTTP, TypeScript SDK, and sync/async Python SDK paths.", + "raw": { + "focalPoints": [ + "production-reproduction", + "sibling-paths", + "response-aggregation" + ], + "adjustments": "Kept the existing byte cap and added the same 200-file ceiling at both durable-outbox and direct client boundaries; left circuit breaker and HTTP contract unchanged.", + "confidence": 0.9 + }, + "significance": "high", + "tags": [ + "focal:production-reproduction", + "focal:sibling-paths", + "focal:response-aggregation", + "confidence:0.9" + ] + } + ] + } + ], + "retrospective": { + "summary": "Fixed #445 by capping bulk write requests at 200 files, covering durable mount outbox maintenance and all public client bulk-write methods; added over-limit red/green regression coverage and verified all Go/TypeScript/Python suites.", + "approach": "Standard approach", + "confidence": 0.9 + }, + "commits": [], + "filesChanged": [], + "projectId": "/Users/khaliqgant/Projects/AgentWorkforce/relayfile/.lanes/relayfile-445-sql-variable-limit", + "tags": [], + "_trace": { + "startRef": "fecb9ce404a47e25d18303d5c4ae80139eba2562", + "endRef": "fecb9ce404a47e25d18303d5c4ae80139eba2562" + } +} \ No newline at end of file diff --git a/.trajectories/completed/2026-08/traj_hhthwa7xxikz.md b/.trajectories/completed/2026-08/traj_hhthwa7xxikz.md new file mode 100644 index 00000000..1000596e --- /dev/null +++ b/.trajectories/completed/2026-08/traj_hhthwa7xxikz.md @@ -0,0 +1,33 @@ +# Trajectory: Fix #445 SQL variable limit data loss + +> **Status:** ✅ Completed +> **Task:** AgentWorkforce/relayfile#445 +> **Confidence:** 90% +> **Started:** August 24, 2026 at 11:12 AM +> **Completed:** August 24, 2026 at 11:35 AM + +--- + +## Summary + +Fixed #445 by capping bulk write requests at 200 files, covering durable mount outbox maintenance and all public client bulk-write methods; added over-limit red/green regression coverage and verified all Go/TypeScript/Python suites. + +**Approach:** Standard approach + +--- + +## Key Decisions + +### Cap bulk writes at 200 files across durable outbox and client boundaries +- **Chose:** Cap bulk writes at 200 files across durable outbox and client boundaries +- **Reasoning:** The outbox maintenance path chunks only by 8 MiB, so 1,454 small records can be reassembled into one request. A fixed 200-file cap stays well below SQLite's conservative 999-variable limit, preserves byte chunking, and protects foreground, maintenance, and direct client callers. + +--- + +## Chapters + +### 1. Work +*Agent: default* + +- Cap bulk writes at 200 files across durable outbox and client boundaries: Cap bulk writes at 200 files across durable outbox and client boundaries +- Reproduced the production signature with 1,001 small workspace files; count chunking is green across Go mount, direct Go HTTP, TypeScript SDK, and sync/async Python SDK paths. diff --git a/.trajectories/index.json b/.trajectories/index.json index 44a37940..de6ff051 100644 --- a/.trajectories/index.json +++ b/.trajectories/index.json @@ -1,6 +1,6 @@ { "version": 1, - "lastUpdated": "2026-08-23T14:45:33.239Z", + "lastUpdated": "2026-08-24T09:35:03.379Z", "trajectories": { "traj_4pvrlmqfnzng": { "title": "Review PR #278 in AgentWorkforce/relayfile", @@ -197,6 +197,13 @@ "startedAt": "2026-08-23T14:42:40.801Z", "completedAt": "2026-08-23T14:45:33.103Z", "path": ".trajectories/completed/2026-08/traj_2ng1fbz1wsxb.json" + }, + "traj_hhthwa7xxikz": { + "title": "Fix #445 SQL variable limit data loss", + "status": "completed", + "startedAt": "2026-08-24T09:12:49.140Z", + "completedAt": "2026-08-24T09:35:03.059Z", + "path": ".trajectories/completed/2026-08/traj_hhthwa7xxikz.json" } } } diff --git a/internal/mountsync/http_client_test.go b/internal/mountsync/http_client_test.go index 063ac17a..b98a9af9 100644 --- a/internal/mountsync/http_client_test.go +++ b/internal/mountsync/http_client_test.go @@ -6,6 +6,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "io" "net/http" "net/http/httptest" @@ -460,3 +461,73 @@ func TestHTTPClientWriteFilesBulkRejectsEmptyBatch(t *testing.T) { t.Fatalf("expected empty batch to fail before issuing a request, got %d calls", atomic.LoadInt32(&calls)) } } + +func TestHTTPClientWriteFilesBulkChunksLargeCallerSlice(t *testing.T) { + const fileCount = 1001 + + var calls atomic.Int32 + var largestBatch atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/workspaces/ws_large_bulk/fs/bulk" { + t.Fatalf("unexpected path %q", r.URL.Path) + } + var payload struct { + Files []BulkWriteFile `json:"files"` + } + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Fatalf("decode request body failed: %v", err) + } + batchSize := int32(len(payload.Files)) + for previous := largestBatch.Load(); batchSize > previous; previous = largestBatch.Load() { + if largestBatch.CompareAndSwap(previous, batchSize) { + break + } + } + call := calls.Add(1) + if len(payload.Files) > 999 { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusInternalServerError) + _ = json.NewEncoder(w).Encode(map[string]string{ + "code": "internal_error", + "message": "too many SQL variables at offset 412: SQLITE_ERROR", + }) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(BulkWriteResponse{ + Written: len(payload.Files), + Results: []BulkWriteResult{{ + Path: payload.Files[0].Path, + Revision: fmt.Sprintf("rev_%d", call), + }}, + CorrelationID: fmt.Sprintf("corr_%d", call), + }) + })) + defer server.Close() + + files := make([]BulkWriteFile, 0, fileCount) + for idx := 0; idx < fileCount; idx++ { + files = append(files, BulkWriteFile{ + Path: fmt.Sprintf("/notion/Docs/File%04d.md", idx+1), + Content: "small", + }) + } + client := NewHTTPClient(server.URL, "token", server.Client()) + response, err := client.WriteFilesBulk(context.Background(), "ws_large_bulk", files) + if err != nil { + t.Fatalf("large bulk write failed: %v", err) + } + if response.Written != fileCount || response.ErrorCount != 0 { + t.Fatalf("unexpected aggregate response: %+v", response) + } + wantCalls := int32((fileCount + maxBulkWriteFilesPerRequest - 1) / maxBulkWriteFilesPerRequest) + if got := calls.Load(); got != wantCalls { + t.Fatalf("bulk request count = %d, want %d", got, wantCalls) + } + if got := len(response.Results); got != int(wantCalls) { + t.Fatalf("aggregate result count = %d, want %d", got, wantCalls) + } + if got := largestBatch.Load(); got > maxBulkWriteFilesPerRequest { + t.Fatalf("largest bulk request = %d, safe cap = %d", got, maxBulkWriteFilesPerRequest) + } +} diff --git a/internal/mountsync/syncer.go b/internal/mountsync/syncer.go index ad5553fb..16642e5d 100644 --- a/internal/mountsync/syncer.go +++ b/internal/mountsync/syncer.go @@ -77,7 +77,15 @@ func (e *SchemaValidationError) Is(target error) bool { return target == ErrSchemaValidation } -const defaultBulkFlushThreshold = 256 +const ( + defaultBulkFlushThreshold = 256 + // maxBulkWriteFilesPerRequest keeps caller-sized /fs/bulk payloads well + // below SQLite's conservative 999-variable build limit. The cloud storage + // layer may bind one variable per file while committing a bulk mutation, so + // the byte-size cap alone is insufficient for large workspaces of small + // files. + maxBulkWriteFilesPerRequest = 200 +) // The WebSocket server may inline file payloads up to 1 MiB. nhooyr's client // default is only 32 KiB, which would disconnect real-time mounts whenever a @@ -755,6 +763,25 @@ func (c *HTTPClient) WriteFilesBulk(ctx context.Context, workspaceID string, fil if len(files) == 0 { return BulkWriteResponse{}, ErrEmptyBulkWrite } + var combined BulkWriteResponse + for start := 0; start < len(files); start += maxBulkWriteFilesPerRequest { + end := min(start+maxBulkWriteFilesPerRequest, len(files)) + response, err := c.writeFilesBulkBatch(ctx, workspaceID, files[start:end]) + if err != nil { + return combined, err + } + combined.Written += response.Written + combined.ErrorCount += response.ErrorCount + combined.Errors = append(combined.Errors, response.Errors...) + combined.Results = append(combined.Results, response.Results...) + if strings.TrimSpace(response.CorrelationID) != "" { + combined.CorrelationID = response.CorrelationID + } + } + return combined, nil +} + +func (c *HTTPClient) writeFilesBulkBatch(ctx context.Context, workspaceID string, files []BulkWriteFile) (BulkWriteResponse, error) { body := struct { Files []BulkWriteFile `json:"files"` }{ @@ -3882,14 +3909,13 @@ func chunkOutboxRecords(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 && bulkWriteRequestSize(outboxRecordsAsBulkFiles(candidate)) > maxBytes { + exceedsFileLimit := len(current) >= maxBulkWriteFilesPerRequest + exceedsByteLimit := maxBytes > 0 && bulkWriteRequestSize(outboxRecordsAsBulkFiles(candidate)) > maxBytes + if len(current) > 0 && (exceedsFileLimit || exceedsByteLimit) { chunks = append(chunks, append([]outboxRecord(nil), current...)) current = current[:0] } diff --git a/internal/mountsync/syncer_test.go b/internal/mountsync/syncer_test.go index 41ec45a7..e6551788 100644 --- a/internal/mountsync/syncer_test.go +++ b/internal/mountsync/syncer_test.go @@ -5510,17 +5510,77 @@ func TestBulkWrite_ChunkAtThreshold(t *testing.T) { t.Fatalf("chunked sync failed: %v", err) } - if client.bulkWriteCalls != 2 { - t.Fatalf("expected two bulk write calls, got %d", client.bulkWriteCalls) + if client.bulkWriteCalls != 3 { + t.Fatalf("expected three bulk write calls, got %d", client.bulkWriteCalls) + } + if got := len(client.bulkWriteBatches); got != 3 { + t.Fatalf("expected three recorded bulk batches, got %d", got) + } + if got := len(client.bulkWriteBatches[0]); got != 200 { + t.Fatalf("expected first batch size 200, got %d", got) + } + if got := len(client.bulkWriteBatches[1]); got != 56 { + t.Fatalf("expected second batch size 56, got %d", got) + } + if got := len(client.bulkWriteBatches[2]); got != 44 { + t.Fatalf("expected third batch size 44, got %d", got) + } +} + +func TestBulkWrite_LargeWorkspaceStaysBelowSQLVariableLimit(t *testing.T) { + const ( + fileCount = 1001 + sqlVariableLimit = 999 + ) + + client := &fakeClient{ + files: map[string]RemoteFile{}, + bulkWriteResponseFunc: func(_ context.Context, _ string, files []BulkWriteFile) (BulkWriteResponse, error) { + if len(files) > sqlVariableLimit { + return BulkWriteResponse{}, &HTTPError{ + StatusCode: http.StatusInternalServerError, + Code: "internal_error", + Message: "too many SQL variables at offset 412: SQLITE_ERROR", + } + } + return BulkWriteResponse{}, nil + }, + } + localDir := t.TempDir() + if err := os.MkdirAll(filepath.Join(localDir, "Docs"), 0o755); err != nil { + t.Fatalf("mkdir docs failed: %v", err) } - if got := len(client.bulkWriteBatches); got != 2 { - t.Fatalf("expected two recorded bulk batches, got %d", got) + for idx := 0; idx < fileCount; idx++ { + name := fmt.Sprintf("File%04d.md", idx+1) + path := filepath.Join(localDir, "Docs", name) + if err := os.WriteFile(path, []byte(name), 0o644); err != nil { + t.Fatalf("seed file %s failed: %v", name, err) + } } - if got := len(client.bulkWriteBatches[0]); got != 256 { - t.Fatalf("expected first batch size 256, got %d", got) + + syncer, err := NewSyncer(client, SyncerOptions{ + WorkspaceID: "ws_mount_sql_variable_limit", + RemoteRoot: "/notion", + LocalRoot: localDir, + }) + if err != nil { + t.Fatalf("new syncer failed: %v", err) } - if got := len(client.bulkWriteBatches[1]); got != 44 { - t.Fatalf("expected second batch size 44, got %d", got) + // Reproduce the maintenance path that can collect every small due outbox + // record into one request instead of stopping at the normal foreground + // discovery threshold. + syncer.bulkFlushThreshold = fileCount + + if err := syncer.SyncOnce(context.Background()); err != nil { + t.Fatalf("large-workspace sync failed: %v", err) + } + if got := len(client.files); got != fileCount { + t.Fatalf("remote file count = %d, want %d", got, fileCount) + } + for idx, batch := range client.bulkWriteBatches { + if len(batch) > maxBulkWriteFilesPerRequest { + t.Fatalf("batch %d contains %d files, over safe cap %d", idx, len(batch), maxBulkWriteFilesPerRequest) + } } } diff --git a/packages/sdk/python/src/relayfile/client.py b/packages/sdk/python/src/relayfile/client.py index e55bdc74..dc8efb99 100644 --- a/packages/sdk/python/src/relayfile/client.py +++ b/packages/sdk/python/src/relayfile/client.py @@ -27,6 +27,7 @@ AccessTokenProvider = Union[str, Callable[[], str]] AsyncAccessTokenProvider = Union[str, Callable[[], Any]] # sync or async callable +MAX_BULK_WRITE_FILES_PER_REQUEST = 200 @dataclass @@ -116,6 +117,33 @@ def _read_payload(response: httpx.Response) -> Any: return {"message": response.text} +def _merge_bulk_write_response( + combined: dict[str, Any] | None, batch: Any +) -> dict[str, Any]: + if combined is None: + return dict(batch) if isinstance(batch, dict) else {} + if not isinstance(batch, dict): + return combined + + merged: dict[str, Any] = { + "written": int(combined.get("written", 0)) + int(batch.get("written", 0)), + "errorCount": int(combined.get("errorCount", 0)) + + int(batch.get("errorCount", 0)), + "errors": [ + *(combined.get("errors") or []), + *(batch.get("errors") or []), + ], + "correlationId": batch.get("correlationId") + or combined.get("correlationId", ""), + } + if combined.get("results") is not None or batch.get("results") is not None: + merged["results"] = [ + *(combined.get("results") or []), + *(batch.get("results") or []), + ] + return merged + + def _parse_retry_after_ms(header: str | None) -> float | None: if not header: return None @@ -487,12 +515,20 @@ def bulk_write( *, correlation_id: str | None = None, ) -> dict[str, Any]: - return self._request( - "POST", - f"/v1/workspaces/{_enc(workspace_id)}/fs/bulk", - json_body={"files": files}, - correlation_id=correlation_id, - ) + batches = [ + files[start : start + MAX_BULK_WRITE_FILES_PER_REQUEST] + for start in range(0, len(files), MAX_BULK_WRITE_FILES_PER_REQUEST) + ] or [[]] + result: dict[str, Any] | None = None + for batch in batches: + batch_result = self._request( + "POST", + f"/v1/workspaces/{_enc(workspace_id)}/fs/bulk", + json_body={"files": batch}, + correlation_id=correlation_id, + ) + result = _merge_bulk_write_response(result, batch_result) + return result or {} def delete_file(self, input: DeleteFileInput) -> dict[str, Any]: query = _build_query({"path": input.path}) @@ -1187,12 +1223,20 @@ async def bulk_write( *, correlation_id: str | None = None, ) -> dict[str, Any]: - return await self._request( - "POST", - f"/v1/workspaces/{_enc(workspace_id)}/fs/bulk", - json_body={"files": files}, - correlation_id=correlation_id, - ) + batches = [ + files[start : start + MAX_BULK_WRITE_FILES_PER_REQUEST] + for start in range(0, len(files), MAX_BULK_WRITE_FILES_PER_REQUEST) + ] or [[]] + result: dict[str, Any] | None = None + for batch in batches: + batch_result = await self._request( + "POST", + f"/v1/workspaces/{_enc(workspace_id)}/fs/bulk", + json_body={"files": batch}, + correlation_id=correlation_id, + ) + result = _merge_bulk_write_response(result, batch_result) + return result or {} async def delete_file(self, input: DeleteFileInput) -> dict[str, Any]: query = _build_query({"path": input.path}) diff --git a/packages/sdk/python/tests/test_client.py b/packages/sdk/python/tests/test_client.py index 4562415c..b30723e4 100644 --- a/packages/sdk/python/tests/test_client.py +++ b/packages/sdk/python/tests/test_client.py @@ -127,6 +127,44 @@ def test_bulk_write(self) -> None: assert req.headers["X-Correlation-Id"] == "corr_bulk" assert json.loads(req.content)["files"][0]["path"] == "/a.md" + @respx.mock + def test_bulk_write_chunks_large_caller_slice(self) -> None: + batch_sizes: list[int] = [] + + def handler(request: httpx.Request) -> httpx.Response: + files = json.loads(request.content)["files"] + batch_sizes.append(len(files)) + if len(files) > 999: + return httpx.Response( + 500, + json={ + "code": "internal_error", + "message": "too many SQL variables at offset 412: SQLITE_ERROR", + }, + ) + return httpx.Response( + 202, + json={ + "written": len(files), + "errorCount": 0, + "errors": [], + "correlationId": f"corr_{len(batch_sizes)}", + }, + ) + + respx.post(f"{BASE}/v1/workspaces/ws_acme/fs/bulk").mock( + side_effect=handler + ) + files = [ + {"path": f"/docs/File{index + 1:04d}.md", "content": "small"} + for index in range(1001) + ] + + result = self._client().bulk_write("ws_acme", files) + + assert result["written"] == len(files) + assert batch_sizes == [200, 200, 200, 200, 200, 1] + @respx.mock def test_query_files(self) -> None: payload = {"items": [], "nextCursor": None} @@ -675,6 +713,38 @@ async def test_bulk_write(self) -> None: ) assert res["written"] == 1 + @respx.mock + @pytest.mark.asyncio + async def test_bulk_write_chunks_large_caller_slice(self) -> None: + batch_sizes: list[int] = [] + + def handler(request: httpx.Request) -> httpx.Response: + files = json.loads(request.content)["files"] + batch_sizes.append(len(files)) + return httpx.Response( + 202, + json={ + "written": len(files), + "errorCount": 0, + "errors": [], + "correlationId": f"corr_{len(batch_sizes)}", + }, + ) + + respx.post(f"{BASE}/v1/workspaces/ws_acme/fs/bulk").mock( + side_effect=handler + ) + files = [ + {"path": f"/docs/File{index + 1:04d}.md", "content": "small"} + for index in range(1001) + ] + + async with AsyncRelayFileClient(BASE, "tok_test") as client: + result = await client.bulk_write("ws_acme", files) + + assert result["written"] == len(files) + assert batch_sizes == [200, 200, 200, 200, 200, 1] + @respx.mock @pytest.mark.asyncio async def test_export_workspace_json(self) -> None: diff --git a/packages/sdk/typescript/src/client.test.ts b/packages/sdk/typescript/src/client.test.ts index 93ff38e0..ce63cd0e 100644 --- a/packages/sdk/typescript/src/client.test.ts +++ b/packages/sdk/typescript/src/client.test.ts @@ -1414,6 +1414,36 @@ describe("RelayFileClient — existing methods", () => { ], }); }); + + it("chunks caller slices that exceed SQLite's conservative variable limit", async () => { + const batchSizes: number[] = []; + const f = vi.fn(async (_url: string | URL | Request, init?: RequestInit) => { + const files = (JSON.parse(init?.body as string) as { files: unknown[] }).files; + batchSizes.push(files.length); + if (files.length > 999) { + return new Response(JSON.stringify({ + code: "internal_error", + message: "too many SQL variables at offset 412: SQLITE_ERROR", + }), { status: 500, headers: { "content-type": "application/json" } }); + } + return new Response(JSON.stringify({ + written: files.length, + errorCount: 0, + errors: [], + correlationId: `corr_${batchSizes.length}`, + }), { status: 202, headers: { "content-type": "application/json" } }); + }); + const client = makeClient(f as typeof fetch); + const files = Array.from({ length: 1001 }, (_, index) => ({ + path: `/docs/File${String(index + 1).padStart(4, "0")}.md`, + content: "small", + })); + + const result = await client.bulkWrite({ workspaceId: "ws_acme", files }); + + expect(result.written).toBe(files.length); + expect(batchSizes).toEqual([200, 200, 200, 200, 200, 1]); + }); }); describe("checkpoint seals", () => { diff --git a/packages/sdk/typescript/src/client.ts b/packages/sdk/typescript/src/client.ts index ea2667ba..95d4ea69 100644 --- a/packages/sdk/typescript/src/client.ts +++ b/packages/sdk/typescript/src/client.ts @@ -102,6 +102,8 @@ import { RevisionConflictError } from "./errors.js"; +const MAX_BULK_WRITE_FILES_PER_REQUEST = 200; + /** * Bearer token or token factory used for Relayfile API requests. * @@ -1645,23 +1647,44 @@ export class RelayFileClient { async bulkWrite(input: BulkWriteInput): Promise { const query = buildQuery({ forkId: input.forkId }); - const response = await this.performRequest({ - method: "POST", - path: `/v1/workspaces/${encodeURIComponent(input.workspaceId)}/fs/bulk${query}`, - correlationId: input.correlationId, - body: { - files: input.files - }, - signal: input.signal - }); - const result = await (this.readPayload(response) as Promise); + const batches: BulkWriteInput["files"][] = []; + for (let start = 0; start < input.files.length; start += MAX_BULK_WRITE_FILES_PER_REQUEST) { + batches.push(input.files.slice(start, start + MAX_BULK_WRITE_FILES_PER_REQUEST)); + } + // Preserve the server's validation behavior for an empty caller slice. + if (batches.length === 0) batches.push([]); + + let result: BulkWriteResponse | undefined; + for (const files of batches) { + const response = await this.performRequest({ + method: "POST", + path: `/v1/workspaces/${encodeURIComponent(input.workspaceId)}/fs/bulk${query}`, + correlationId: input.correlationId, + body: { files }, + signal: input.signal + }); + const batchResult = await (this.readPayload(response) as Promise); + if (!result) { + result = batchResult; + continue; + } + result = { + written: result.written + batchResult.written, + errorCount: result.errorCount + batchResult.errorCount, + errors: [...result.errors, ...batchResult.errors], + ...(result.results || batchResult.results + ? { results: [...(result.results ?? []), ...(batchResult.results ?? [])] } + : {}), + correlationId: batchResult.correlationId || result.correlationId + }; + } const cache = getFileReadCache(this); if (cache !== false) { for (const file of input.files) { cache.evict(input.workspaceId, file.path); } } - return result; + return result!; } async issueCheckpointSeal(input: IssueCheckpointSealInput): Promise {