Skip to content

Commit e43a169

Browse files
fix(storage): S3 chunked uploads resolve their object key from the upload session row (#31705)
<!-- claude-projects-attribution: {"github_login":"os-steve"} --> _Requested by **Steve** · [project thread](https://claude.ai/code/project/chan_01FdL1XeUyvVNxLreh3F4Qi9?thread=cmsg_01FdL1XeUyvVNxLreh3F4Qi9XSBUAWKhacmmrYT4zZC5MZ)_ Fixes #31607 Clause-②: no Metadata-protocol: none ## 维护者速读 - 改了什么:S3 分片上传的分片门与完成门在调用后端前,从持久的 `sys_upload_session.key` 取对象 key 交给适配器;可替换存储代理补上 `setUploadKey` 转发。 - 为什么改:经代理时适配器的进程内 uploadId→key 表从未被填,单副本上第 0 片就 500;这张表只在一个进程里,换副本必然找不到。 - 风险与回滚:只动 S3 适配器经过的路径;本地适配器与无 `setUploadKey` 的适配器调用不变(各一个 pin)。回滚即 revert 本 PR,无数据迁移。 - 先看:`storage-routes.ts`(`handUploadKeyToBackend` 及分片门、完成门两处调用)、`swappable-storage-service.ts`(`setUploadKey` 转发)、新测试 `s3-chunked-upload-session-key.test.ts`。 - 两分钟验证:`pnpm --filter @objectstack/service-storage exec vitest run src/s3-chunked-upload-session-key.test.ts`,预期 4/4 通过;在 main 上同一文件两个 S3 用例 500 失败。 - 需要你做的:无,席位落地。 ## What changes - `storage-routes.ts`: a new module helper `handUploadKeyToBackend(storage, uploadId, key)` hands the backend the object key of a chunked upload. The chunk door and the completion door call it with `session.key` (the `sys_upload_session` row each door already reads) right before `uploadChunk` / `completeChunkedUpload`. The init door no longer calls `setUploadKey` itself: every backend call that names the upload now gets the key from the row, on whichever process serves it. - `swappable-storage-service.ts`: the proxy forwards `setUploadKey` to the active adapter, and is a no-op for an adapter without it (same pattern as the proxy's existing `verifyToken` and `getPresignedUploadOrigin` passthroughs). This is what made chunk 0 fail on a single replica: the kernel's `storage` service is the proxy, and the init door's `'setUploadKey' in storage` probe never matched it. - `s3-storage-adapter.ts`, `attachment-lifecycle.ts`: comments only. They described the map as filled once by the plugin, and said the proxy does not forward `setUploadKey`. Both were made false by this change. No code change in either file. - `.changeset/31607-s3-chunked-upload-session-key.md`: `@objectstack/service-storage` patch. No change to the `IStorageService` contract (`packages/spec`), to the chunked-upload HTTP requests or answers, to `sys_upload_session` (it already stores `key`, a required field), or to the local adapter (it records the key beside its parts at initiation). Why the row, not a new key-taking method on the adapter: the card prefers the durable row, and the row already holds the key, so no column is added. Handing it down through the existing `setUploadKey` keeps a single way to address an upload on the adapter; the upload-session reap guard already re-seeds the same map from the same row before an abort, so the chunk and completion doors now follow that precedent. No new public name is introduced on the package entry; the proxy gains one method that the S3 adapter already had. ## Tests New file `packages/services/service-storage/src/s3-chunked-upload-session-key.test.ts` (4 cases). It drives the real doors (`registerStorageRoutes`) through `SwappableStorageService`, the way `StorageServicePlugin` composes them, against an S3-compatible stand-in. Like S3, the stand-in refuses a part, completion or abort addressed to an UploadId under a different key (`NoSuchUpload`). - `completes on a single replica`: init, chunk 0, chunk 1 and completion through one proxy and one S3 adapter; the bucket holds the whole file and no multipart is left open. - `resumes on a fresh adapter instance: another process finds the key in the session row`: three replicas (separate adapter instances behind separate proxies) share one metadata store and one bucket. Init runs on A, chunk 0 on B, chunk 1 and completion on C. The object lands under the session row's key and the file row is `committed`. - `local: the assembled file holds the bytes sent, across two adapter instances on one root`: unchanged behaviour for the local adapter. - `an adapter with no setUploadKey receives exactly the contract calls, with the arguments it did before`: an in-memory adapter double sees the same four calls with the same arguments as on `main`. A new file, not a new `describe` in an existing harness: the S3 stand-in is a `vi.mock` of `@aws-sdk/client-s3`, which is file-scoped, and no existing file mocks S3 and mounts the routes together. No existing test was changed. <details><summary>Step 0: the defect on main (5d1382a), new file against the unfixed code</summary> ``` [5xx] 500 — S3StorageAdapter: key not found for uploadId. Call setUploadKey() before uploadChunk(). × completes on a single replica × resumes on a fresh adapter instance: another process finds the key in the session row AssertionError: {"success":false,"error":{"code":"INTERNAL","message":"Internal error"}}: expected 500 to be 200 Tests 2 failed | 2 passed (4) ``` `sys_upload_session` fields written by the init door: `id` (= the backend UploadId when the backend initiated), `file_id`, `key` (required), `filename`, `mime_type`, `total_size`, `chunk_size`, `total_chunks`, `resume_token`, `backend_upload_id`, `scope`, `bucket`, `metadata`, `status`, `expires_at`; the chunk door adds `parts`, `uploaded_chunks`, `uploaded_size`. The object key is present, so no column is needed. </details> <details><summary>Ablations (scripts/ablation-replace.mjs, restore proven blob == HEAD and git diff HEAD empty)</summary> The test imports `./storage-routes.js` and `./swappable-storage-service.js` relatively, so vitest runs `src`; no `dist` leg applies. 1. Back to the in-process table: the init door seeds the map once through the proxy (`ABLATION-INIT-SEED`, count 1) and the chunk and completion doors no longer hand down the row's key (`ABLATION-NO-ROW-KEY`, count 2, anchor 2 → 0). Only the second-process case goes red, with the original error: ``` [5xx] 500 — S3StorageAdapter: key not found for uploadId. Call setUploadKey() before uploadChunk(). × resumes on a fresh adapter instance: another process finds the key in the session row Tests 1 failed | 3 passed (4) ablation-replace: ok restored: blob == HEAD (8e21af9) and `git diff HEAD` is empty ``` 2. The proxy drops the key (`ABLATION-PROXY-DROPS-KEY`, anchor 1 → 0). Both S3 cases go red: ``` × completes on a single replica × resumes on a fresh adapter instance: another process finds the key in the session row Tests 2 failed | 2 passed (4) ablation-replace: ok restored: blob == HEAD (a5b6010) and `git diff HEAD` is empty ``` </details> ## Acceptance notes - The S3 adapter's map still gains an entry on every process that serves a chunk, and only the process that completes or aborts the upload deletes its entry. An upload abandoned before completion already left its entry behind on `main`. The entries are small and are not a record anything trusts, since the row is the record. Noted, not filed. - The upload-session reap guard (`createUploadSessionReapGuard`) still reaches the inner adapter through `getInner()` to seed the key before an abort. That still works, and the proxy now forwards the call too. It was left as is so that its tests are not changed. ## Verification All on head `14e4eea9`. - `pnpm --filter @objectstack/service-storage build && … test && … typecheck` under the lock: VERDICT command-exit 0; Test Files 76 passed (76), Tests 1428 passed (1428); `check:test-typecheck` OK. - Lint: root `eslint --no-inline-config packages/services/service-storage/src` (the package has no `lint` script), exit 0. - Derived gate families: `node scripts/pm/dispatch-gates.mjs --commands --repo objectstack-ai/objectstack` listed 73; all 73 ran after `turbo run build --filter='./packages/**' --filter='./examples/**'` (VERDICT command-exit 0), every one exit 0. `--ran` reconciliation: 73 derived, 73 run, 0 NOT-MEASURED, 0 UNRUN. - Dogfood, targeted (no `!` in the title, no `(narrowing)` arm, so the full battery is CI's): `storage-chunked-parallel-parts`, `storage-chunked-resume-integrity`, `attachments-permission-matrix`, `error-sentences-live` under the lock: VERDICT command-exit 0; Test Files 4 passed (4), Tests 37 passed, 1 skipped (38). These run the local adapter, so they show the local path unchanged; the S3 path is covered by the new unit file. - `origin/main` moved 12 commits past the base `5d1382a5`; none touches `packages/services/service-storage`, so no merge was taken. ## Human Minimum - [x] 1. Schemas start as Zod; `packages/spec` holds no runtime logic. (`packages/spec` untouched.) - [x] 2. Every name I added uses the casing for its kind. - [x] 3. No formatter ran; `pnpm lint` passes (root eslint `--no-inline-config` over `packages/services/service-storage/src`, exit 0). - [x] 4. One issue, one branch off an up-to-date `main`, one pull request. - [x] 5. A change that publishes carries a changeset, and both declaration lines are filled in here and in the changeset. - [x] 6. `packages/spec` touched: not touched. - [x] 7. No hand edits to `content/docs/references/`, `packages/*/CHANGELOG.md` or `content/docs/releases/`. - [x] 8. Title, body and comments are in English (the 维护者速读 section is the lane's Chinese summary for the maintainer). - [x] The *Before Submitting* list in CONTRIBUTING.md is done. Card thread session: `session_01ADHkmRmL5ZWYTvnS9e3kmo` (dev, `domain:server#4`). 🤖 Generated with [Claude Code](https://claude.com/claude-code) https://claude.ai/code/session_01ADHkmRmL5ZWYTvnS9e3kmo --- _Generated by [Claude Code](https://claude.ai/code/session_01ADHkmRmL5ZWYTvnS9e3kmo)_ Co-authored-by: Claude <noreply@anthropic.com>
1 parent f127b04 commit e43a169

6 files changed

Lines changed: 374 additions & 17 deletions

File tree

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
---
2+
'@objectstack/service-storage': patch
3+
---
4+
5+
fix(storage): S3 chunked uploads find their object key in the upload session row, on any replica
6+
7+
Clause-②: no
8+
Metadata-protocol: none
9+
10+
Chunked uploads to S3 work again. Who is affected: any deployment whose storage adapter is S3 (or S3-compatible) and whose clients upload in chunks (`POST /storage/upload/chunked`, the SDK's chunked and resumable uploads). Before this fix the first chunk answered `500` ("S3StorageAdapter: key not found for uploadId") even on a single server, and a chunk or completion served by another replica could never have found the key.
11+
12+
S3 addresses a multipart upload by object key and upload id, but the storage contract's chunk and completion calls carry the upload id alone, so the S3 adapter kept the key in an in-memory map filled at the start of the upload. The kernel's `storage` service is the swappable proxy, which did not forward that call, so the map was never filled; and a map in one process cannot serve another. The chunk and completion doors now read the key from `sys_upload_session.key` (the durable record the init door already writes) and hand it to the adapter before every backend call, and the proxy forwards `setUploadKey` to the active adapter.
13+
14+
Nothing changes for the local adapter or any adapter without `setUploadKey`, nor in the chunked-upload HTTP requests and answers. No migration: upload sessions already store their key.

‎packages/services/service-storage/src/attachment-lifecycle.ts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -763,8 +763,9 @@ export function createUploadSessionReapGuard(
763763
// A cold sweep runs long after the live session, so the S3 adapter's
764764
// in-process `uploadId → key` map (populated by `setUploadKey` during
765765
// upload) is empty — re-seed it from the row so the abort can resolve
766-
// the S3 key. `setUploadKey` is S3-specific and not forwarded by the
767-
// swappable proxy, so reach the inner adapter; a no-op for `local`.
766+
// the S3 key. `setUploadKey` is S3-specific and off the IStorageService
767+
// contract; the inner adapter is reached directly (the swappable proxy
768+
// forwards it too); a no-op for `local`.
768769
const inner: any = typeof (storage as any).getInner === 'function' ? (storage as any).getInner() : storage;
769770
if (typeof row.key === 'string' && row.key && typeof inner?.setUploadKey === 'function') {
770771
inner.setUploadKey(backendId, row.key);
Lines changed: 298 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,298 @@
1+
// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
2+
3+
/**
4+
* A chunked upload to S3 finds its object key in the upload session row, on
5+
* any process.
6+
*
7+
* S3 addresses a multipart upload by `(Key, UploadId)`, but the
8+
* `IStorageService` chunk and completion methods carry the `uploadId` alone.
9+
* The S3 adapter answers the key from its own `uploadId → key` map, which
10+
* `setUploadKey` fills. Two ways that map came up empty:
11+
*
12+
* 1. The plugin registers the SWAPPABLE proxy as the kernel's `storage`
13+
* service, and the proxy did not forward `setUploadKey`, so the init
14+
* door's call never reached the adapter. Chunk 0 answered 500 on a single
15+
* replica.
16+
* 2. The map lives in one process. A chunk or completion served by another
17+
* replica (or by the same replica after a restart or an adapter swap)
18+
* had no entry, whatever the init door did.
19+
*
20+
* The durable record of the key is `sys_upload_session.key`, written by the
21+
* init door. The chunk and completion doors read that row before they reach
22+
* the backend, so they hand its key to the adapter on every call. These cases
23+
* drive the doors against an S3-compatible stand-in that, like S3, refuses a
24+
* part or a completion addressed to the wrong key.
25+
*/
26+
27+
import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
28+
29+
const fakeS3 = vi.hoisted(() => {
30+
const objects = new Map<string, Buffer>();
31+
const multipart = new Map<string, { key: string; parts: Map<number, Buffer> }>();
32+
let uploadSeq = 0;
33+
34+
function noSuchUpload(): Error {
35+
const err = new Error('NoSuchUpload') as Error & { name: string; $metadata: { httpStatusCode: number } };
36+
err.name = 'NoSuchUpload';
37+
err.$metadata = { httpStatusCode: 404 };
38+
return err;
39+
}
40+
41+
/** S3 matches BOTH halves of the address: an UploadId under another Key is no upload at all. */
42+
function upload(input: { Key?: string; UploadId?: string }) {
43+
const found = input.UploadId ? multipart.get(input.UploadId) : undefined;
44+
if (!found || found.key !== input.Key) throw noSuchUpload();
45+
return found;
46+
}
47+
48+
function reset(): void {
49+
objects.clear();
50+
multipart.clear();
51+
uploadSeq = 0;
52+
}
53+
54+
return { objects, multipart, upload, reset, nextUploadId: () => `s3-upload-${++uploadSeq}` };
55+
});
56+
57+
vi.mock('@aws-sdk/client-s3', () => {
58+
class Command {
59+
constructor(public readonly input: Record<string, any>) {}
60+
}
61+
class CreateMultipartUploadCommand extends Command {}
62+
class UploadPartCommand extends Command {}
63+
class CompleteMultipartUploadCommand extends Command {}
64+
class AbortMultipartUploadCommand extends Command {}
65+
66+
class S3Client {
67+
constructor(_config: unknown) {}
68+
async send(command: any): Promise<any> {
69+
const input = command.input;
70+
if (command instanceof CreateMultipartUploadCommand) {
71+
const uploadId = fakeS3.nextUploadId();
72+
fakeS3.multipart.set(uploadId, { key: input.Key, parts: new Map() });
73+
return { UploadId: uploadId };
74+
}
75+
if (command instanceof UploadPartCommand) {
76+
fakeS3.upload(input).parts.set(input.PartNumber, input.Body);
77+
return { ETag: `"etag-${input.PartNumber}"` };
78+
}
79+
if (command instanceof CompleteMultipartUploadCommand) {
80+
const found = fakeS3.upload(input);
81+
const listed: Array<{ PartNumber: number }> = input.MultipartUpload.Parts;
82+
fakeS3.objects.set(input.Key, Buffer.concat(listed.map((p) => found.parts.get(p.PartNumber)!)));
83+
fakeS3.multipart.delete(input.UploadId);
84+
return {};
85+
}
86+
if (command instanceof AbortMultipartUploadCommand) {
87+
fakeS3.upload(input);
88+
fakeS3.multipart.delete(input.UploadId);
89+
return {};
90+
}
91+
throw new Error(`S3 stand-in: unhandled command ${command.constructor.name}`);
92+
}
93+
}
94+
95+
return {
96+
S3Client,
97+
CreateMultipartUploadCommand,
98+
UploadPartCommand,
99+
CompleteMultipartUploadCommand,
100+
AbortMultipartUploadCommand,
101+
};
102+
});
103+
104+
import type { IHttpRequest, IHttpResponse, IStorageService, RouteHandler } from '@objectstack/spec/contracts';
105+
import { promises as fs } from 'node:fs';
106+
import { join } from 'node:path';
107+
import { tmpdir } from 'node:os';
108+
import { S3StorageAdapter } from './s3-storage-adapter.js';
109+
import { LocalStorageAdapter } from './local-storage-adapter.js';
110+
import { SwappableStorageService } from './swappable-storage-service.js';
111+
import { StorageMetadataStore } from './metadata-store.js';
112+
import { registerStorageRoutes } from './storage-routes.js';
113+
114+
const BASE = '/api/v1/storage';
115+
const INIT = `${BASE}/upload/chunked`;
116+
const CHUNK = `${BASE}/upload/chunked/:uploadId/chunk/:chunkIndex`;
117+
const COMPLETE = `${BASE}/upload/chunked/:uploadId/complete`;
118+
119+
/** The smallest chunk the init door accepts (5 MiB), so two chunks are one full and one short. */
120+
const CHUNK_SIZE = 5 * 1024 * 1024;
121+
const FIRST = Buffer.alloc(CHUNK_SIZE, 0x61);
122+
const SECOND = Buffer.from('tail');
123+
124+
/**
125+
* The stored object is the whole file, byte for byte. Compared with
126+
* `Buffer.equals` rather than `toEqual`, which walks 5 MiB one element at a time.
127+
*/
128+
function expectWhole(stored: Buffer | undefined): void {
129+
expect(stored?.length).toBe(FIRST.length + SECOND.length);
130+
expect(stored?.equals(Buffer.concat([FIRST, SECOND]))).toBe(true);
131+
}
132+
133+
type Res = IHttpResponse & { _status: number; _json: any };
134+
135+
function createMockRes(): Res {
136+
const res: any = {
137+
_status: 200,
138+
_json: null,
139+
json(data: any) { res._json = data; },
140+
send(data: any) { res._sent = data; },
141+
status(code: number) { res._status = code; return res; },
142+
header() { return res; },
143+
};
144+
return res;
145+
}
146+
147+
function createMockReq(overrides: Record<string, unknown>): IHttpRequest {
148+
return { params: {}, query: {}, body: undefined, headers: {}, method: 'GET', path: '/', ...overrides } as IHttpRequest;
149+
}
150+
151+
/**
152+
* One server process: its own adapter instance (so its own in-process state)
153+
* mounted behind its own swappable proxy, the way `StorageServicePlugin`
154+
* composes it. Replicas share the metadata store (the database) and the bucket.
155+
*/
156+
function replica(store: StorageMetadataStore, inner: IStorageService) {
157+
const routes = new Map<string, RouteHandler>();
158+
const server = {
159+
get: (path: string, h: RouteHandler) => { routes.set(`GET:${path}`, h); },
160+
post: (path: string, h: RouteHandler) => { routes.set(`POST:${path}`, h); },
161+
put: (path: string, h: RouteHandler) => { routes.set(`PUT:${path}`, h); },
162+
delete: () => {},
163+
patch: () => {},
164+
use: () => {},
165+
};
166+
registerStorageRoutes(server as any, new SwappableStorageService(inner), store, { basePath: BASE });
167+
const call = async (method: string, path: string, req: Record<string, unknown>): Promise<Res> => {
168+
const res = createMockRes();
169+
await routes.get(`${method}:${path}`)!(createMockReq(req), res);
170+
return res;
171+
};
172+
return {
173+
init: () =>
174+
call('POST', INIT, {
175+
body: { filename: 'big.bin', mimeType: 'application/octet-stream', totalSize: FIRST.length + SECOND.length, chunkSize: CHUNK_SIZE },
176+
}),
177+
chunk: (uploadId: string, resumeToken: string, chunkIndex: number, bytes: Buffer) =>
178+
call('PUT', CHUNK, {
179+
params: { uploadId, chunkIndex: String(chunkIndex) },
180+
headers: { 'x-resume-token': resumeToken },
181+
rawBody: async () => bytes,
182+
}),
183+
complete: (uploadId: string, parts: Array<{ chunkIndex: number; eTag: string }>) =>
184+
call('POST', COMPLETE, { params: { uploadId }, body: { uploadId, parts } }),
185+
};
186+
}
187+
188+
const s3 = () => new S3StorageAdapter({ bucket: 'uploads', region: 'us-east-1', keyPrefix: null });
189+
190+
/** Drive one upload: init on `a`, chunk 0 on `b`, chunk 1 and completion on `c`. */
191+
async function drive(a: ReturnType<typeof replica>, b: ReturnType<typeof replica>, c: ReturnType<typeof replica>) {
192+
const init = await a.init();
193+
expect(init._status, JSON.stringify(init._json)).toBe(200);
194+
const { uploadId, resumeToken } = init._json.data;
195+
196+
const first = await b.chunk(uploadId, resumeToken, 0, FIRST);
197+
expect(first._status, JSON.stringify(first._json)).toBe(200);
198+
const second = await c.chunk(uploadId, resumeToken, 1, SECOND);
199+
expect(second._status, JSON.stringify(second._json)).toBe(200);
200+
201+
const done = await c.complete(uploadId, [
202+
{ chunkIndex: 0, eTag: first._json.data.eTag },
203+
{ chunkIndex: 1, eTag: second._json.data.eTag },
204+
]);
205+
expect(done._status, JSON.stringify(done._json)).toBe(200);
206+
return done._json.data as { fileId: string; key: string; size: number };
207+
}
208+
209+
beforeEach(() => {
210+
fakeS3.reset();
211+
});
212+
213+
describe('a chunked upload through the swappable proxy reaches the S3 object it started', () => {
214+
it('completes on a single replica', async () => {
215+
const one = replica(new StorageMetadataStore(null), s3());
216+
217+
const done = await drive(one, one, one);
218+
219+
expect(done.size).toBe(FIRST.length + SECOND.length);
220+
expectWhole(fakeS3.objects.get(done.key));
221+
expect(fakeS3.multipart.size).toBe(0);
222+
});
223+
224+
it('resumes on a fresh adapter instance: another process finds the key in the session row', async () => {
225+
// One database and one bucket; three processes, none sharing memory. The
226+
// replica that serves a chunk or the completion has never seen the init.
227+
const store = new StorageMetadataStore(null);
228+
const a = replica(store, s3());
229+
const b = replica(store, s3());
230+
const c = replica(store, s3());
231+
232+
const done = await drive(a, b, c);
233+
234+
expectWhole(fakeS3.objects.get(done.key));
235+
expect(fakeS3.multipart.size).toBe(0);
236+
// The key the object landed under is the one the session row recorded.
237+
const file = await store.getFile(done.fileId);
238+
expect(file?.key).toBe(done.key);
239+
expect(file?.status).toBe('committed');
240+
});
241+
});
242+
243+
describe('adapters that do not track upload keys are untouched', () => {
244+
let rootDir: string;
245+
246+
afterEach(async () => {
247+
if (rootDir) await fs.rm(rootDir, { recursive: true, force: true });
248+
});
249+
250+
it('local: the assembled file holds the bytes sent, across two adapter instances on one root', async () => {
251+
rootDir = join(tmpdir(), `os-chunk-session-key-${Date.now()}-${Math.random().toString(36).slice(2)}`);
252+
await fs.mkdir(rootDir, { recursive: true });
253+
const store = new StorageMetadataStore(null);
254+
const local = () => new LocalStorageAdapter({ rootDir, signingSecret: 'test-secret' });
255+
const reader = local();
256+
257+
const done = await drive(replica(store, local()), replica(store, local()), replica(store, local()));
258+
259+
expectWhole(await reader.download(done.key));
260+
});
261+
262+
it('an adapter with no setUploadKey receives exactly the contract calls, with the arguments it did before', async () => {
263+
const calls: unknown[][] = [];
264+
const parts = new Map<string, Buffer[]>();
265+
const memory: IStorageService = {
266+
upload: async () => {},
267+
download: async () => Buffer.alloc(0),
268+
delete: async () => {},
269+
exists: async () => false,
270+
getInfo: async (key) => ({ key, size: 0, lastModified: new Date(0) }),
271+
initiateChunkedUpload: async (key) => {
272+
calls.push(['initiateChunkedUpload', key]);
273+
parts.set('mem-1', []);
274+
return 'mem-1';
275+
},
276+
uploadChunk: async (uploadId, partNumber, data) => {
277+
calls.push(['uploadChunk', uploadId, partNumber, data.length]);
278+
parts.get(uploadId)!.push(data);
279+
return `m${partNumber}`;
280+
},
281+
completeChunkedUpload: async (uploadId, listed) => {
282+
calls.push(['completeChunkedUpload', uploadId, listed]);
283+
return 'assembled-key';
284+
},
285+
};
286+
287+
const one = replica(new StorageMetadataStore(null), memory);
288+
const done = await drive(one, one, one);
289+
290+
expect(done.key).toBe('assembled-key');
291+
expect(calls).toEqual([
292+
['initiateChunkedUpload', expect.stringMatching(/^user\/.+\.bin$/)],
293+
['uploadChunk', 'mem-1', 1, FIRST.length],
294+
['uploadChunk', 'mem-1', 2, SECOND.length],
295+
['completeChunkedUpload', 'mem-1', [{ partNumber: 1, eTag: 'm1' }, { partNumber: 2, eTag: 'm2' }]],
296+
]);
297+
});
298+
});

‎packages/services/service-storage/src/s3-storage-adapter.ts‎

Lines changed: 15 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -607,13 +607,11 @@ export class S3StorageAdapter implements IStorageService {
607607
async uploadChunk(uploadId: string, partNumber: number, data: Buffer): Promise<string> {
608608
const client = await this.getClient();
609609
const s3 = await this.s3Mod();
610-
// We need the key — store the relationship elsewhere or pass via metadata.
611-
// For the S3 adapter, `uploadId` is the S3-native UploadId. The key is
612-
// tracked in the StorageMetadataStore (sys_upload_session.key).
613-
// Here we retrieve it from session state; the plugin ensures the correct
614-
// key is passed. However, the IStorageService contract doesn't include key
615-
// in uploadChunk — so we work around by storing the mapping in a WeakMap
616-
// keyed by uploadId. For a robust implementation we'll add a lookup:
610+
// S3 addresses a part by key AND UploadId, but the IStorageService
611+
// contract carries the UploadId alone. The durable record of the key is
612+
// `sys_upload_session.key`; the storage routes hand it down through
613+
// `setUploadKey` before every chunk and completion call, on whichever
614+
// process serves the request.
617615
const key = this._uploadKeys?.get(uploadId);
618616
if (!key) {
619617
throw new Error('S3StorageAdapter: key not found for uploadId. Call setUploadKey() before uploadChunk().');
@@ -672,10 +670,16 @@ export class S3StorageAdapter implements IStorageService {
672670
private _uploadKeys: Map<string, string> = new Map();
673671

674672
/**
675-
* Register the storage key for a multipart upload session. Must be called
676-
* by the StorageServicePlugin after `initiateChunkedUpload()` returns so
677-
* that subsequent `uploadChunk` / `completeChunkedUpload` calls can resolve
678-
* the S3 key without it being part of the IStorageService contract signature.
673+
* Register the storage key (the CALLER key, unprefixed) for a multipart
674+
* upload, so `uploadChunk` / `completeChunkedUpload` / `abortChunkedUpload`
675+
* can address it without the key being part of the IStorageService contract
676+
* signature.
677+
*
678+
* This map is a per-process hand-off, never the record: the storage routes
679+
* call this (through the swappable proxy, which forwards it) with
680+
* `sys_upload_session.key` before every chunk and completion call, and the
681+
* upload-session reap guard does the same before an abort, so a request
682+
* served by a process that never saw the upload start still finds its key.
679683
*/
680684
setUploadKey(uploadId: string, key: string): void {
681685
this._uploadKeys.set(uploadId, key);

0 commit comments

Comments
 (0)