From b981898776014cf29408af47be4362f8cb6efcfe Mon Sep 17 00:00:00 2001 From: Peter Gaultney Date: Mon, 17 Aug 2026 15:51:28 -0400 Subject: [PATCH] fix: retry attachment uploads and heal content missing from storage An attachment syncs in two decoupled halves: the metadata (path + sha256) rides the folder doc over CRDT sync and always converges, while the content is a client PUT to storage. When that PUT died with no HTTP response, the classifier had no code or status to match, fell back to the generic message, and treated the failure as terminal - the background queue only re-drives retryable failures, so the upload was abandoned permanently. The metadata landed anyway, leaving every other client and any git-sync chasing a hash that exists nowhere. We found six such attachments in one vault, the oldest three weeks old and never retried across many restarts. - CAS.writeFile gains the withTransientRetry wrapper readFile already had. - Any non-abort transfer failure with no HTTP response is retryable (s3TransferFailureFromUnknown); DOMException AbortError and abort-like messages stay terminal. - Existing holes heal: each attachment version is HEAD-verified against storage once per machine, bounded by a durable guid -> hash ledger (casVerified), and a machine whose local content IS the missing version re-uploads it. A machine holding a different version surfaces an error instead of clobbering the claim - the bytes may still exist on the author's machine. Successful pushes and pulls record for free. - sweepUnverifiedCasContent enqueues the verification after each syncFileTree. Without it the verify never runs for files at rest: the tree sync noops any file whose local state matches committed metadata, and SyncFile.sync() otherwise only fires on local modify events and in-memory retry queues. - 404 now reads 'Attachment content not found in storage' rather than the generic 'Attachment storage request failed.' enableVerifyUploads is deleted rather than defaulted on. The plugin materializes every flag into data.json on save, so an existing install never consults the schema default - the flag reads false on every machine that has saved settings, and a fix gated on it does nothing for them. The verify path it guarded also called push() without force, whose hash-unchanged guard skipped the upload, so the feature was a no-op for exactly the author-machine case it existed for; that is now push(true). --- package-lock.json | 22 ---- src/CAS.ts | 64 +++++----- src/S3Error.ts | 28 +++++ src/SharedFolder.ts | 40 +++++++ src/SyncFile.ts | 69 +++++++++-- src/flags.ts | 8 -- tests-th/CAS.upload-retry.test.ts | 112 ++++++++++++++++++ tests-th/README.md | 6 + .../S3Error.transfer-classification.test.ts | 50 ++++++++ 9 files changed, 326 insertions(+), 73 deletions(-) create mode 100644 tests-th/CAS.upload-retry.test.ts create mode 100644 tests-th/README.md create mode 100644 tests-th/S3Error.transfer-classification.test.ts diff --git a/package-lock.json b/package-lock.json index 6e372003e..7d97b3e8a 100644 --- a/package-lock.json +++ b/package-lock.json @@ -5665,21 +5665,6 @@ } } }, - "node_modules/svelte/node_modules/@typescript-eslint/types": { - "version": "8.65.0", - "resolved": "https://registry.npmjs.org/@typescript-eslint/types/-/types-8.65.0.tgz", - "integrity": "sha512-JSSwWNy+H0E/01jJEM+hrX6N0OFDzFzeIhHFSAS01tlVaevpG8cFyYRPhS5yjGOvBUx3sqQHVMjCL1CAZZMxBg==", - "license": "MIT", - "optional": true, - "peer": true, - "engines": { - "node": "^18.18.0 || ^20.9.0 || >=21.1.0" - }, - "funding": { - "type": "opencollective", - "url": "https://opencollective.com/typescript-eslint" - } - }, "node_modules/svelte/node_modules/esrap": { "version": "2.3.0", "resolved": "https://registry.npmjs.org/esrap/-/esrap-2.3.0.tgz", @@ -9980,13 +9965,6 @@ "zimmerframe": "^1.1.2" }, "dependencies": { - "@typescript-eslint/types": { - "version": "8.65.0", - "resolved": "https://registry.npmjs.org/@typescript-eslint/types/-/types-8.65.0.tgz", - "integrity": "sha512-JSSwWNy+H0E/01jJEM+hrX6N0OFDzFzeIhHFSAS01tlVaevpG8cFyYRPhS5yjGOvBUx3sqQHVMjCL1CAZZMxBg==", - "optional": true, - "peer": true - }, "esrap": { "version": "2.3.0", "resolved": "https://registry.npmjs.org/esrap/-/esrap-2.3.0.tgz", diff --git a/src/CAS.ts b/src/CAS.ts index 973b53e98..8bbb34bfe 100644 --- a/src/CAS.ts +++ b/src/CAS.ts @@ -10,6 +10,7 @@ import { s3ApiErrorFromResponse, s3ApiErrorFromUnknown, s3NetworkFailureFromUnknown, + s3TransferFailureFromUnknown, } from "./S3Error"; // In-attempt retry schedule for transient transfer failures: one retry per @@ -120,7 +121,9 @@ export class ContentAddressedStore extends HasLogging { return await request(); } catch (error) { const classified = - s3NetworkFailureFromUnknown(error, operation) ?? error; + s3NetworkFailureFromUnknown(error, operation) ?? + s3TransferFailureFromUnknown(error, operation) ?? + error; const delayCapMs = this.transferRetryDelaysMs[attempt]; if (!isRetryableS3Error(classified) || delayCapMs === undefined) { throw classified; @@ -143,36 +146,37 @@ export class ContentAddressedStore extends HasLogging { if (!(content && hash)) { throw new Error("invalid caf"); } - const token = await this.tokenStore.getFileToken( - S3RN.encode(syncFile.s3rn), - hash, - syncFile.mimetype, - content.byteLength, - ); - const response = await customFetch(token.baseUrl + "/upload-url", { - method: "POST", - headers: { Authorization: `Bearer ${token.token}` }, - relayNetworkDomain: "relay", + return this.withTransientRetry("upload attachment", async () => { + const token = await this.tokenStore.getFileToken( + S3RN.encode(syncFile.s3rn), + hash, + syncFile.mimetype, + content.byteLength, + ); + const response = await customFetch(token.baseUrl + "/upload-url", { + method: "POST", + headers: { Authorization: `Bearer ${token.token}` }, + relayNetworkDomain: "relay", + }); + if (response.status !== 200) { + throw await this.s3ResponseError(response, "upload attachment url"); + } + const responseJson = await response.json(); + const presignedUrl = responseJson.uploadUrl; + const uploadResponse = await this.s3Request( + () => + customFetch(presignedUrl, { + method: "PUT", + headers: { "Content-Type": syncFile.mimetype }, + body: content, + relayNetworkDomain: "external", + }), + "upload attachment", + ); + if (!uploadResponse.ok) { + throw await this.s3ResponseError(uploadResponse, "upload attachment"); + } }); - if (response.status !== 200) { - throw await this.s3ResponseError(response, "upload attachment url"); - } - const responseJson = await response.json(); - const presignedUrl = responseJson.uploadUrl; - const uploadResponse = await this.s3Request( - () => - customFetch(presignedUrl, { - method: "PUT", - headers: { "Content-Type": syncFile.mimetype }, - body: content, - relayNetworkDomain: "external", - }), - "upload attachment", - ); - if (!uploadResponse.ok) { - throw await this.s3ResponseError(uploadResponse, "upload attachment"); - } - return; } private async s3Request( diff --git a/src/S3Error.ts b/src/S3Error.ts index 7f2ad3706..738ccb1d2 100644 --- a/src/S3Error.ts +++ b/src/S3Error.ts @@ -85,6 +85,31 @@ export function s3NetworkFailureFromUnknown( const NETWORK_FAILURE_PATTERN = /net::ERR_|Failed to fetch|fetch failed|\bLoad failed\b|NetworkError|network error|socket hang up|ECONNRESET|ECONNREFUSED|ECONNABORTED|ETIMEDOUT|ENOTFOUND|EPIPE|EAI_AGAIN/i; +/** + * Classify a failure thrown mid-transfer when no HTTP response was produced. + * Unlike the pattern-matched s3NetworkFailureFromUnknown, any non-abort + * failure is treated as transport-level and retryable: a transfer that + * produced no response cannot have been refused by policy, and parking it + * strands the file (the background queue only re-drives retryable failures). + * Aborts stay null so a cancelled transfer is not re-driven by the retry + * loop that was just cancelled. + */ +export function s3TransferFailureFromUnknown( + error: unknown, + operation?: string, +): S3ApiError | null { + if (error instanceof S3ApiError) return null; + if (error instanceof DOMException && error.name === "AbortError") { + return null; + } + const text = errorText(error) ?? ""; + if (/abort/i.test(text)) return null; + return new S3ApiError( + { code: "NetworkingError", operation, message: text || undefined }, + error, + ); +} + function jsonErrorMessage(body: string): string | undefined { try { const parsed = JSON.parse(body) as { error?: unknown }; @@ -159,6 +184,9 @@ function userMessageForS3Error(details: S3ErrorDetails): string { if (details.status === 429) { return "Attachment storage is busy. Relay will retry the upload."; } + if (details.status === 404) { + return "Attachment content not found in storage."; + } if (details.status !== undefined && details.status >= 500) { return "Attachment storage is temporarily unavailable. Relay will retry the upload."; } diff --git a/src/SharedFolder.ts b/src/SharedFolder.ts index 8b5e99804..95fb6e09b 100644 --- a/src/SharedFolder.ts +++ b/src/SharedFolder.ts @@ -241,6 +241,7 @@ export class SharedFolder extends HasProvider { private syncRequestedDuringSync: boolean = false; private authoritative: boolean; private pendingUpload: LocalStorage; + private casVerified: LocalStorage; private unsubscribes: Unsubscriber[] = []; private storageQuota?: number; /** @@ -386,6 +387,9 @@ export class SharedFolder extends HasProvider { this.pendingUpload = new LocalStorage( `${appId}-system3-relay/folders/${this.guid}/pendingUploads`, ); + this.casVerified = new LocalStorage( + `${appId}-system3-relay/folders/${this.guid}/casVerified`, + ); this.pendingUpload.forEach((guid, vpath) => { if (!this.existsSync(vpath)) { this.warn( @@ -3287,6 +3291,27 @@ export class SharedFolder extends HasProvider { this.warn("dropped stale pending-upload holds", stale); } + /** + * Enqueue a sync for every attachment version not yet in the casVerified + * ledger. The tree sync noops files whose local state matches committed + * metadata, so their sync() — where remote-content verification lives — + * otherwise never runs for at-rest files. The ledger bounds this to one + * verification per (file, hash) per machine. + */ + private sweepUnverifiedCasContent(): void { + for (const file of this.files.values()) { + if (!isSyncFile(file) || file.destroyed) continue; + + const meta = this.syncStore.getMeta(file.path); + if (!meta || !isFileMetas(meta) || !meta.hash) continue; + if (this.isCasVerified(file.guid, meta.hash)) continue; + + this.backgroundSync.enqueueSync(file).catch((error) => { + this.warn(`cas verification sync failed for ${file.path}`, error); + }); + } + } + syncFileTree(): Promise { // If a sync is already running, mark that we want another sync after if (this.syncFileTreePromise) { @@ -3375,6 +3400,7 @@ export class SharedFolder extends HasProvider { this.log("syncFileTree diff:\n" + diffLog.join("\n")); } this.sweepStalePendingUploads(); + this.sweepUnverifiedCasContent(); } finally { // Reset the promise after completion (success or failure) this.syncFileTreePromise = null; @@ -3566,6 +3592,20 @@ export class SharedFolder extends HasProvider { this.pendingUpload.clear(); } + /** + * Durable per-machine ledger of attachment versions whose content is + * known to exist in storage, so remote-existence verification runs at + * most once per (file, hash) instead of on every sync pass. Content is + * immutable under its hash, so an entry never needs re-checking. + */ + public isCasVerified(guid: string, hash: string): boolean { + return this.casVerified.get(guid) === hash; + } + + public markCasVerified(guid: string, hash: string): void { + this.casVerified.set(guid, hash); + } + async markUploaded( file: IFile, outcome: SyncCompletionOutcome = "completed", diff --git a/src/SyncFile.ts b/src/SyncFile.ts index b7e5dc023..50ce376ad 100644 --- a/src/SyncFile.ts +++ b/src/SyncFile.ts @@ -13,7 +13,6 @@ import { Observable, type Unsubscriber } from "./observable/Observable"; import { generateHash } from "./hashing"; import type { HasMimeType, IFile } from "./IFile"; import { getMimeType } from "./mimetypes"; -import { flags } from "./flagManager"; import { errorFromUnknown, formatUserFacingError } from "./UserFacingError"; export function isSyncFile(file: IFile | undefined): file is SyncFile { @@ -557,6 +556,9 @@ export class SyncFile try { await this.sharedFolder.cas.writeFile(this); await this.sharedFolder.markUploaded(this); + if (hash) { + this.sharedFolder.markCasVerified(this.guid, hash); + } this.uploadError = undefined; this.notifyListeners(); this.debug("push complete", { @@ -648,7 +650,10 @@ export class SyncFile return; } - if (this.isCleanLastServerEdit(this.meta as FileMetas, this.stat)) { + if ( + this.isCleanLastServerEdit(this.meta as FileMetas, this.stat) && + this.sharedFolder.isCasVerified(this.guid, this.meta.hash) + ) { this.debug("sync decision", { path: this.path, guid: this.guid, @@ -672,17 +677,7 @@ export class SyncFile statMtime: this.stat.mtime, metaSynctime: this.meta.synctime, }); - if (flags().enableVerifyUploads) { - // Not remote - try { - if (!(await this.verifyUpload())) { - this.warn("file in metadata, but not on the server!"); - await this.push(); - } - } catch (err) { - // pass - } - } + await this.reconcileRemoteContent(hash); if (hash === this.meta.hash) { this.clearCurrentUserEdit(); this.debug("sync decision", { @@ -795,6 +790,53 @@ export class SyncFile return this.sharedFolder.cas.verify(this); } + /** + * Confirm the version named by metadata actually exists in storage, + * at most once per (file, hash). A hole is healable only from a machine + * whose local content IS the missing version; a machine holding a + * different version must not clobber the claim - the missing bytes may + * still exist on the authoring machine. + */ + private async reconcileRemoteContent(localHash: string | null) { + if (!this.meta) { + return; + } + const metaHash = this.meta.hash; + if (this.sharedFolder.isCasVerified(this.guid, metaHash)) { + return; + } + let exists: boolean; + try { + exists = await this.verifyUpload(); + } catch (err) { + this.debug( + "remote content verification failed; retrying on a later sync", + err, + ); + return; + } + if (exists) { + this.sharedFolder.markCasVerified(this.guid, metaHash); + return; + } + if (localHash && localHash === metaHash) { + this.warn( + `[${this.path}] content missing from storage; re-uploading local copy`, + ); + await this.push(true); + return; + } + this.warn( + `[${this.path}] content missing from storage and the local copy is a different version; only the authoring machine can heal this`, + ); + const message = + "Attachment content was never uploaded; awaiting its author's machine."; + if (this.uploadError !== message) { + this.uploadError = message; + this.notifyListeners(); + } + } + public async pull() { this.log("pull"); this._refreshMeta(); @@ -843,6 +885,7 @@ export class SyncFile .catch((error) => { this.warn("Failed to save pulled hash:", error); }); + this.sharedFolder.markCasVerified(this.guid, this.meta.hash); if (this.uploadError) { this.uploadError = undefined; this.notifyListeners(); diff --git a/src/flags.ts b/src/flags.ts index a273906d6..2fadc74fc 100644 --- a/src/flags.ts +++ b/src/flags.ts @@ -4,7 +4,6 @@ export interface FeatureFlags { enableDiffLinkStatus: boolean; enableDeltaLogging: boolean; enableNetworkLogging: boolean; - enableVerifyUploads: boolean; enableDiscordLogin: boolean; enableDeviceManagement: boolean; enableHSMRecording: boolean; @@ -104,13 +103,6 @@ export const FeatureFlagSchema: { description: "Log HTTP status, method, URL, and response bodies from Relay network requests.", }, - enableVerifyUploads: { - default: false, - category: "debugging", - title: "Verify uploaded attachments", - description: - "After attachment sync, confirm the remote object exists and re-upload if it is missing.", - }, enableHSMRecording: { default: false, category: "debugging", diff --git a/tests-th/CAS.upload-retry.test.ts b/tests-th/CAS.upload-retry.test.ts new file mode 100644 index 000000000..5da1c1d88 --- /dev/null +++ b/tests-th/CAS.upload-retry.test.ts @@ -0,0 +1,112 @@ +jest.mock("src/customFetch", () => ({ customFetch: jest.fn() })); +jest.mock("pocketbase", () => ({ + __esModule: true, + default: jest.fn().mockImplementation(() => ({ + cancelAllRequests: jest.fn(), + })), +})); +jest.mock("src/S3RN", () => ({ + S3RN: { encode: jest.fn(() => "s3rn:relay:test:doc:test") }, +})); + +import { ContentAddressedStore } from "src/CAS"; +import { customFetch } from "src/customFetch"; +import type { SyncFile } from "src/SyncFile"; +import type { SharedFolder } from "src/SharedFolder"; + +const mockFetch = customFetch as jest.Mock; + +function makeStore(): ContentAddressedStore { + const sharedFolder = { + path: "test-folder", + loginManager: { + getEndpointManager: () => ({ getAuthUrl: () => "https://auth.example" }), + authStore: {}, + }, + tokenStore: { + getFileToken: jest.fn(async () => ({ + baseUrl: "https://relay.example/f/doc", + token: "tok", + })), + }, + } as unknown as SharedFolder; + return new ContentAddressedStore(sharedFolder, { + transferRetryDelaysMs: [0, 0], + }); +} + +function makeSyncFile(): SyncFile { + return { + caf: { + read: async () => new TextEncoder().encode("svg bytes").buffer, + hash: async () => "abc123", + }, + s3rn: {}, + mimetype: "image/svg+xml", + guid: "test-guid", + } as unknown as SyncFile; +} + +function okJson(payload: object): Response { + return { + ok: true, + status: 200, + json: async () => payload, + } as unknown as Response; +} + +function okPut(): Response { + return { ok: true, status: 200 } as unknown as Response; +} + +beforeEach(() => { + mockFetch.mockReset(); +}); + +describe("writeFile retry", () => { + it("retries an upload whose transport died with an unrecognized error", async () => { + mockFetch + .mockRejectedValueOnce( + new TypeError("some transport failure no pattern knows about"), + ) + .mockResolvedValueOnce(okJson({ uploadUrl: "https://s3.example/put" })) + .mockResolvedValueOnce(okPut()); + await expect(makeStore().writeFile(makeSyncFile())).resolves.toBeUndefined(); + expect(mockFetch).toHaveBeenCalledTimes(3); + }); + + it("retries when the presigned PUT itself dies mid-flight", async () => { + mockFetch + .mockResolvedValueOnce(okJson({ uploadUrl: "https://s3.example/put" })) + .mockRejectedValueOnce(new Error("socket hang up")) + .mockResolvedValueOnce(okJson({ uploadUrl: "https://s3.example/put" })) + .mockResolvedValueOnce(okPut()); + await expect(makeStore().writeFile(makeSyncFile())).resolves.toBeUndefined(); + expect(mockFetch).toHaveBeenCalledTimes(4); + }); + + it("does not retry an aborted upload", async () => { + mockFetch.mockRejectedValue( + new DOMException("The operation was aborted.", "AbortError"), + ); + await expect(makeStore().writeFile(makeSyncFile())).rejects.toThrow(); + expect(mockFetch).toHaveBeenCalledTimes(1); + }); + + it("does not retry a policy refusal", async () => { + mockFetch.mockResolvedValue({ + ok: false, + status: 403, + text: async () => "", + } as unknown as Response); + await expect(makeStore().writeFile(makeSyncFile())).rejects.toThrow(); + expect(mockFetch).toHaveBeenCalledTimes(1); + }); + + it("gives up after the retry schedule is exhausted", async () => { + mockFetch.mockRejectedValue(new Error("ECONNRESET")); + await expect(makeStore().writeFile(makeSyncFile())).rejects.toThrow(); + // initial attempt + one per schedule entry ([0, 0] -> 3 total) + expect(mockFetch).toHaveBeenCalledTimes(3); + }); +}); diff --git a/tests-th/README.md b/tests-th/README.md new file mode 100644 index 000000000..0d3cc3841 --- /dev/null +++ b/tests-th/README.md @@ -0,0 +1,6 @@ +Tests contributed from outside the git-crypt key. `__tests__/**` is encrypted, so a +contributor without the key can neither read its conventions nor add a reviewable file +there; these live in plaintext instead. Jest's default testMatch picks them up +(`npx jest tests-th`). + +Maintainers are welcome to move these under `__tests__/` and delete this directory. diff --git a/tests-th/S3Error.transfer-classification.test.ts b/tests-th/S3Error.transfer-classification.test.ts new file mode 100644 index 000000000..6a79c5f22 --- /dev/null +++ b/tests-th/S3Error.transfer-classification.test.ts @@ -0,0 +1,50 @@ +import { + S3ApiError, + isRetryableS3Error, + s3ApiErrorFromResponse, + s3TransferFailureFromUnknown, +} from "src/S3Error"; + +describe("s3TransferFailureFromUnknown", () => { + it("classifies an unrecognized no-response failure as retryable", () => { + const classified = s3TransferFailureFromUnknown( + new TypeError("some transport failure no pattern knows about"), + "upload attachment", + ); + expect(classified).toBeInstanceOf(S3ApiError); + expect(isRetryableS3Error(classified)).toBe(true); + expect(classified?.code).toBe("NetworkingError"); + }); + + it("classifies a non-Error throw as retryable", () => { + const classified = s3TransferFailureFromUnknown("connection dropped", "upload attachment"); + expect(isRetryableS3Error(classified)).toBe(true); + }); + + it("returns null for an AbortError DOMException", () => { + expect( + s3TransferFailureFromUnknown( + new DOMException("The operation was aborted.", "AbortError"), + ), + ).toBeNull(); + }); + + it("returns null for abort-shaped messages", () => { + expect( + s3TransferFailureFromUnknown(new Error("The user aborted a request.")), + ).toBeNull(); + }); + + it("never re-wraps an already-classified S3ApiError", () => { + const original = s3ApiErrorFromResponse(403, "", "upload attachment"); + expect(s3TransferFailureFromUnknown(original)).toBeNull(); + }); +}); + +describe("404 user message", () => { + it("names missing content instead of the generic fallback", () => { + const error = s3ApiErrorFromResponse(404, "", "download attachment url"); + expect(error.message).toBe("Attachment content not found in storage."); + expect(error.retryable).toBe(false); + }); +});