diff --git a/.changeset/curly-planets-lead.md b/.changeset/curly-planets-lead.md index 890d4c36b..8f0b2cbeb 100644 --- a/.changeset/curly-planets-lead.md +++ b/.changeset/curly-planets-lead.md @@ -3,4 +3,4 @@ --- Add attachments support via `TanStackDBAttachmentQueue`. This extends the PowerSync SDK's `AttachmentQueue` and backs it with -a TanStack DB collection, so file uploads/deletes are managed atomically alongside the relational data. +a TanStack DB collection, so attachment metadata and related rows commit atomically. Local files and remote uploads/deletes are managed separately. diff --git a/.changeset/powersync-attachment-startup-ownership.md b/.changeset/powersync-attachment-startup-ownership.md new file mode 100644 index 000000000..d30adce3d --- /dev/null +++ b/.changeset/powersync-attachment-startup-ownership.md @@ -0,0 +1,5 @@ +--- +'@tanstack/powersync-db-collection': patch +--- + +Load attachment IDs before save/delete in eager and on-demand collections. Preserve existing files when a duplicate save is rejected, reject overlapping saves of the same ID across queues sharing a database, and clean up partial local writes. diff --git a/docs/collections/powersync-collection.md b/docs/collections/powersync-collection.md index 7982a8f37..0ffaa382b 100644 --- a/docs/collections/powersync-collection.md +++ b/docs/collections/powersync-collection.md @@ -1103,7 +1103,7 @@ const liveQuery = createLiveQueryCollection({ ## Attachments -`@tanstack/powersync-db-collection` ships `TanStackDBAttachmentQueue`, an [`AttachmentQueue`](https://docs.powersync.com/usage/use-case-examples/attachments-files) whose file operations commit inside a TanStack DB collection transaction. This lets you create (or delete) an attachment and mutate a related collection row (for example, setting `lists.photo_id`) atomically in a single transaction, instead of issuing two independent writes. +`@tanstack/powersync-db-collection` ships `TanStackDBAttachmentQueue`, an [`AttachmentQueue`](https://docs.powersync.com/usage/use-case-examples/attachments-files) that commits attachment metadata and related collection mutations (for example, setting `lists.photo_id`) in one database transaction. File I/O is separate: a failed save attempts to remove its local file, while the SDK performs remote uploads and deletes later. The queue extends PowerSync's `AttachmentQueue`, so the generic concepts are unchanged and documented once in the SDK. @@ -1130,6 +1130,10 @@ These are standard PowerSync attachment requirements. See the SDK attachments do This is the piece that makes the integration TanStack-aware: a normal PowerSync collection over the attachments table. The queue reads and writes attachment records through it. +Both eager and on-demand collections work. Before `save` or `delete` opens its mutation, the queue loads the attachment ID through a temporary live query and retains that query until the transaction is confirmed. It does not call `preload()` inside a mutation function or require loading the entire table. + +An existing ID, or a concurrent save of that ID through the same PowerSync database object, is rejected before writing the file. This is an in-process guard, not a lock across separate database handles, SDK queues, tabs, or processes. File names retain the SDK's ID-based convention so restart can find files after the app's storage directory moves. + ```ts import { createCollection } from "@tanstack/react-db" import { powerSyncCollectionOptions } from "@tanstack/powersync-db-collection" @@ -1235,6 +1239,8 @@ await attachmentQueue.save({ `delete` queues the file for deletion and runs your `updateHook` in the same transaction. Clear the foreign key so the row and the attachment stay consistent. As with `save`, the hook must be synchronous. +**Upstream limitation:** the SDK version used by this PR can overwrite a queued deletion when an already-running upload succeeds or fails. The related row is detached, but the SDK can lose the remote deletion or retry the obsolete upload. This integration does not work around that SDK completion race. The [attachment oracle notes](../../packages/powersync-db-collection/tests/ATTACHMENT-ORACLE.md) include runnable native-SDK and integration repros; a green ordinary suite does not establish safety for this overlap. + ```ts await attachmentQueue.delete({ id: photo_id, @@ -1266,4 +1272,4 @@ const { data } = useLiveQuery((q) => attachment_local_uri: attachment?.local_uri, })) ) -``` \ No newline at end of file +``` diff --git a/packages/powersync-db-collection/package.json b/packages/powersync-db-collection/package.json index 5f41c617f..476fb3a50 100644 --- a/packages/powersync-db-collection/package.json +++ b/packages/powersync-db-collection/package.json @@ -27,7 +27,8 @@ "build": "vite build", "dev": "vite build --watch", "lint": "eslint . --fix", - "test": "vitest --run" + "test": "vitest --run", + "test:upstream-repros": "vitest run --config tests/upstream.config.ts --maxWorkers=2" }, "type": "module", "main": "dist/cjs/index.cjs", diff --git a/packages/powersync-db-collection/src/attachments.ts b/packages/powersync-db-collection/src/attachments.ts index 7fb7dd926..465d45d13 100644 --- a/packages/powersync-db-collection/src/attachments.ts +++ b/packages/powersync-db-collection/src/attachments.ts @@ -1,5 +1,9 @@ -import { AttachmentQueue, AttachmentState } from '@powersync/common' -import { createTransaction } from '@tanstack/db' +import { + AttachmentQueue, + AttachmentState, + sanitizeSQL, +} from '@powersync/common' +import { createLiveQueryCollection, createTransaction, eq } from '@tanstack/db' import { PowerSyncTransactor } from './PowerSyncTransactor' import type { @@ -11,6 +15,10 @@ import type { import type { Collection, Transaction } from '@tanstack/db' import type { OptionalExtractedTable } from './helpers' +// SDK context locks belong to individual queues. Reserve saves across queues +// sharing this database so a rejected insert cannot remove another save's file. +const savingIds = new WeakMap>() + export type TanStackDBAttachmentQueueOptions = AttachmentQueueOptions & { /** * For TanStack, we want access to the synced TanStackDB collection. @@ -28,7 +36,8 @@ export interface SaveOptions { /** * Optional custom ID. If not provided, a UUID will be generated. * - * Rejected if an attachment with this ID is already in the queue. + * Rejected if this ID is already in the queue or is being saved by another + * call sharing the same PowerSync database object. */ id?: string /** @@ -89,35 +98,80 @@ export class TanStackDBAttachmentQueue extends AttachmentQueue { const resolvedId = id ?? (await this.generateAttachmentId()) const filename = `${resolvedId}.${fileExtension}` const localUri = this.localStorage.getLocalUri(filename) + let pending = savingIds.get(this.powersync) + if (!pending) savingIds.set(this.powersync, (pending = new Set())) + if (pending.has(resolvedId)) { + throw new Error(`Attachment with id ${resolvedId} is already being saved`) + } + pending.add(resolvedId) + + try { + return await this.withLoadedAttachment(resolvedId, () => + this.withAttachmentContext(async (ctx) => { + // A missing in-memory row is not proof that SQLite has no attachment. + if ( + this.collection.get(resolvedId) || + (await ctx.db.getOptional( + sanitizeSQL`SELECT id FROM ${ctx.tableName} WHERE id = ?`, + [resolvedId], + )) + ) { + throw new Error(`Attachment with id ${resolvedId} already exists`) + } + + try { + const size = await this.localStorage.saveFile(localUri, data) + const attachment: AttachmentQueueRow = { + id: resolvedId, + filename, + media_type: mediaType ?? null, + local_uri: localUri, + state: AttachmentState.QUEUED_UPLOAD, + has_synced: 0, + size, + timestamp: new Date().getTime(), + meta_data: metaData ?? null, + } + + const tanStackDBTransaction = createTransaction({ + autoCommit: false, + mutationFn: async ({ transaction }) => { + await new PowerSyncTransactor({ + database: ctx.db, + }).applyTransaction(transaction) + }, + }) + + await this.runInTransaction(tanStackDBTransaction, () => { + this.collection.insert(attachment) + // allow the user to associate values in this transaction + updateHook?.(attachment) + }) + return attachment + } catch (error) { + /** + * The file is written before the transaction opens, so a failed transaction would + * otherwise leave an orphaned file behind that no attachment record points to. + */ + await this.deleteLocalFile(localUri) + throw error + } + }), + ) + } finally { + pending.delete(resolvedId) + } + } - return this.withAttachmentContext(async (ctx) => { - /** - * Checked before the file is written. Writing first would overwrite the existing - * attachment's local file, and the cleanup below would then delete it — leaving the - * pre-existing record pointing at a file that no longer exists. - * - * Deliberately outside the `try`: this throw must not reach the cleanup, because the file - * at `localUri` belongs to the pre-existing attachment rather than to this call. - */ - if (this.collection.get(resolvedId)) { - throw new Error(`Attachment with id ${resolvedId} already exists`) - } - - const size = await this.localStorage.saveFile(localUri, data) - - const attachment: AttachmentQueueRow = { - id: resolvedId, - filename, - media_type: mediaType ?? null, - local_uri: localUri, - state: AttachmentState.QUEUED_UPLOAD, - has_synced: 0, - size, - timestamp: new Date().getTime(), - meta_data: metaData ?? null, - } - - try { + /** + * Queues a file for deletion from local and remote storage. + * + * Exposes an `updateHook` option which is called inside a TanStackDB transaction, + * relational associations with the provided attachment ID should be cleaned up in this hook. + */ + async delete({ id, updateHook }: DeleteOptions): Promise { + await this.withLoadedAttachment(id, () => + this.withAttachmentContext(async (ctx) => { const tanStackDBTransaction = createTransaction({ autoCommit: false, mutationFn: async ({ transaction }) => { @@ -128,55 +182,41 @@ export class TanStackDBAttachmentQueue extends AttachmentQueue { }) await this.runInTransaction(tanStackDBTransaction, () => { - this.collection.insert(attachment) + const attachment = this.collection.get(id) + if (!attachment) { + throw new Error(`Attachment with id ${id} not found`) + } + + this.collection.update(id, (draft) => { + draft.state = AttachmentState.QUEUED_DELETE + draft.has_synced = 0 + }) + // allow the user to associate values in this transaction updateHook?.(attachment) }) - } catch (error) { - /** - * The file is written before the transaction opens, so a failed transaction would - * otherwise leave an orphaned file behind that no attachment record points to. - */ - await this.deleteLocalFile(localUri) - throw error - } - - return attachment - }) + }), + ) } - /** - * Queues a file for deletion from local and remote storage. - * - * Exposes an `updateHook` option which is called inside a TanStackDB transaction, - * relational associations with the provided attachment ID should be cleaned up in this hook. - */ - async delete({ id, updateHook }: DeleteOptions): Promise { - await this.withAttachmentContext(async (ctx) => { - const tanStackDBTransaction = createTransaction({ - autoCommit: false, - mutationFn: async ({ transaction }) => { - await new PowerSyncTransactor({ - database: ctx.db, - }).applyTransaction(transaction) - }, - }) - - await this.runInTransaction(tanStackDBTransaction, () => { - const attachment = this.collection.get(id) - if (!attachment) { - throw new Error(`Attachment with id ${id} not found`) - } - - this.collection.update(id, (draft) => { - draft.state = AttachmentState.QUEUED_DELETE - draft.has_synced = 0 - }) - - // allow the user to associate values in this transaction - updateHook?.(attachment) - }) + private async withLoadedAttachment( + id: string, + operation: () => Promise, + ): Promise { + const query = createLiveQueryCollection({ + query: (q) => + q + .from({ attachment: this.collection }) + .where(({ attachment }) => eq(attachment.id, id)), }) + try { + // Acquire just this ID before opening a mutation and retain its demand + // until PowerSync has confirmed the transaction back to the collection. + await query.preload() + return await operation() + } finally { + await query.cleanup() + } } /** diff --git a/packages/powersync-db-collection/tests/ATTACHMENT-ORACLE.md b/packages/powersync-db-collection/tests/ATTACHMENT-ORACLE.md new file mode 100644 index 000000000..bf5a2e9c4 --- /dev/null +++ b/packages/powersync-db-collection/tests/ATTACHMENT-ORACLE.md @@ -0,0 +1,37 @@ +# Attachment lifecycle oracle + +The model records user intent (absent, present, deleted) and accepted bytes. It +does not mirror the SDK's attachment-state machine. Real SQLite transactions, +TanStack collection delivery, local files, and SDK completion writes run under +the tests. Only remote upload completion and failure are controlled. + +Each command checks owner references, SQL/collection convergence, accepted file +bytes, and cleanup of every file destination touched by save. Draining checks +remote cleanup and an idle extra tick. Rejected hooks and duplicate saves must +not change the accepted reference or bytes. + +The ordinary suite includes a committed corpus, a fixed fast-check campaign, +and a fresh random campaign. To replay a generated failure, run the matching +test with `POWERSYNC_ATTACHMENT_ORACLE_SEED` and, if supplied by fast-check, +`POWERSYNC_ATTACHMENT_ORACLE_PATH`. + +## Upstream completion limitation + +The SDK currently writes the captured upload record after remote I/O without +preserving a newer `QUEUED_DELETE`. Successful upload can lose the remote delete; +failed upload can restore an obsolete upload for retry. The ordinary generated +suite excludes only deletion while upload is in flight. This is a known gap in +supported behavior, not proof that every lifecycle is correct. + +The same model and fixture retain both desired-contract histories and a full +generated schedule in `attachments-sdk-completion.repro.ts`. Run them explicitly: + +```sh +pnpm --filter @tanstack/powersync-db-collection test:upstream-repros +``` + +These are real failing assertions, not expected failures or skipped assertions. +They are separate from the normal gate because the SDK fix is upstream. Once +completion preserves newer intent, move these histories into the normal corpus +and enable in-flight deletion in its generator. Do not weaken the oracle to +accept a detached owner with leaked remote bytes or a resumed obsolete upload. diff --git a/packages/powersync-db-collection/tests/attachments-lifecycle-fixture.ts b/packages/powersync-db-collection/tests/attachments-lifecycle-fixture.ts new file mode 100644 index 000000000..8653e2d70 --- /dev/null +++ b/packages/powersync-db-collection/tests/attachments-lifecycle-fixture.ts @@ -0,0 +1,467 @@ +import { randomUUID } from 'node:crypto' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { fc } from '@fast-check/vitest' +import { AttachmentTable, Schema, Table, column } from '@powersync/common' +import { NodeFileSystemAdapter, PowerSyncDatabase } from '@powersync/node' +import { createCollection } from '@tanstack/db' +import pDefer from 'p-defer' +import { expect, vi } from 'vitest' +import { powerSyncCollectionOptions } from '../src' +import { TanStackDBAttachmentQueue } from '../src/attachments' +import { TEST_DATABASE_IMPLEMENTATION } from './test-db-implementation' +import type { RemoteStorageAdapter } from '@powersync/common' + +const schema = new Schema({ + owners: new Table({ photo_id: column.text }), + attachments: new AttachmentTable(), +}) +const attachmentId = `photo` +const ownerId = `owner` + +type Command = + | `save` + | `reject-save` + | `duplicate-warm` + | `duplicate-cold` + | `delete` + | `reject-delete` + | `start-upload` + | `finish-upload` + | `fail-upload` + | `drain` + +export interface History { + name: string + commands: Array + bytes: Array +} + +// This is a user-intent model, not an AttachmentState transition table. An ID +// owns bytes only after save commits. Rejected operations leave intent intact. +// Transport completion cannot alter the most recent committed reference. +interface Model { + intent: `absent` | `present` | `deleted` + bytes: Array +} + +interface AttachmentRow { + id: string + local_uri: string | null + size: number | null + state: number + has_synced: number +} + +async function setupOracle() { + const directory = await mkdtemp(join(tmpdir(), `ps-attachment-oracle-`)) + const db = new PowerSyncDatabase({ + database: { + dbFilename: `${randomUUID()}.sqlite`, + dbLocation: directory, + implementation: TEST_DATABASE_IMPLEMENTATION, + }, + schema, + }) + await db.disconnectAndClear() + const local = new NodeFileSystemAdapter(join(directory, `files`)) + await local.initialize() + const attachments = createCollection( + powerSyncCollectionOptions({ + database: db, + table: schema.props.attachments, + }), + ) + const owners = createCollection( + powerSyncCollectionOptions({ database: db, table: schema.props.owners }), + ) + await Promise.all([attachments.stateWhenReady(), owners.stateWhenReady()]) + + // Only the remote transport is controlled. Local bytes, transactions, SQL, + // collection delivery, and SDK completion writes all use the real adapters. + const remote = new Map>() + const io = { uploads: 0, deletes: 0, downloads: 0 } + let uploadGate: + | { + entered: ReturnType> + outcome: ReturnType> + } + | undefined + let syncing: Promise | undefined + const transport: RemoteStorageAdapter = { + async uploadFile(data, attachment) { + io.uploads++ + const captured = Array.from(new Uint8Array(data)) + const gate = uploadGate + if (gate) { + gate.entered.resolve() + if (!(await gate.outcome.promise)) + throw new Error(`injected upload failure`) + } + remote.set(attachment.id, captured) + }, + deleteFile(attachment) { + io.deletes++ + remote.delete(attachment.id) + return Promise.resolve() + }, + downloadFile() { + io.downloads++ + throw new Error(`download is outside this save/delete history`) + }, + } + const queue = new TanStackDBAttachmentQueue({ + db, + attachmentsCollection: attachments, + localStorage: local, + remoteStorage: transport, + // These histories exercise explicit save/delete intent. Watcher lifecycle + // has its own boundary tests; no periodic timer can race our schedule. + watchAttachments: () => {}, + archivedCacheLimit: 0, + }) + // Track actual destinations, including writes whose transaction later fails. + // A fixed ID-derived filename would miss orphaned call-owned files. + const writtenFiles = new Set() + const saveFile = local.saveFile.bind(local) + vi.spyOn(local, `saveFile`).mockImplementation((path, data) => { + writtenFiles.add(path) + return saveFile(path, data) + }) + let uri: string | undefined + + async function finishUpload(succeeds: boolean) { + if (!uploadGate || !syncing) throw new Error(`no upload in flight`) + uploadGate.outcome.resolve(succeeds) + await syncing + syncing = undefined + uploadGate = undefined + } + + async function assertObserved(model: Model, label: string, drained: boolean) { + const rows = await db.getAll( + `SELECT id, local_uri, size, state, has_synced FROM attachments ORDER BY id`, + ) + const references = await db.getAll<{ id: string; photo_id: string | null }>( + `SELECT id, photo_id FROM owners ORDER BY id`, + ) + // SQL is authoritative; collection convergence is an additional assertion, + // never the source of the oracle's expected reference or expected bytes. + await vi.waitFor( + () => { + expect( + attachments.toArray.map( + ({ id, local_uri, size, state, has_synced }) => ({ + id, + local_uri, + size, + state, + has_synced, + }), + ), + label, + ).toEqual(rows) + expect( + owners.toArray.map(({ id, photo_id }) => ({ id, photo_id })), + label, + ).toEqual(references) + }, + { timeout: 3000, interval: 10 }, + ) + const exists = uri !== undefined && (await local.fileExists(uri)) + for (const path of writtenFiles) { + if ( + path !== uri || + model.intent === `absent` || + (drained && model.intent === `deleted`) + ) { + expect( + await local.fileExists(path), + `${label}: unexpected file ${path}`, + ).toBe(false) + } + } + const observed = { + references, + localBytes: exists + ? Array.from(new Uint8Array(await local.readFile(uri!))) + : null, + remoteBytes: remote.get(attachmentId) ?? null, + rows, + } + const evidence = `${label}; I/O=${JSON.stringify(io)}` + expect(references, label).toEqual( + model.intent === `absent` + ? [] + : [ + { + id: ownerId, + photo_id: model.intent === `present` ? attachmentId : null, + }, + ], + ) + // This is a logical FK: PowerSync's synced table views do not enforce a + // SQLite FOREIGN KEY. An attached owner must still resolve to its own row. + if (model.intent === `present`) { + expect(observed, evidence).toMatchObject({ + localBytes: model.bytes, + rows: [{ id: attachmentId, local_uri: uri, size: model.bytes.length }], + }) + if (drained) expect(observed.remoteBytes, label).toEqual(model.bytes) + } else if (model.intent === `absent` || drained) { + expect(observed, evidence).toMatchObject({ + localBytes: null, + remoteBytes: null, + rows: [], + }) + } + // If remote bytes exist at an intermediate boundary, they must come from + // the accepted save, never the rejected duplicate (same size, other bytes). + if (observed.remoteBytes) + expect(observed.remoteBytes, label).toEqual(model.bytes) + } + + async function run(command: Command, model: Model) { + switch (command) { + case `save`: + case `reject-save`: { + const saving = queue.save({ + id: attachmentId, + fileExtension: `bin`, + data: new Uint8Array(model.bytes).buffer, + updateHook: (attachment) => { + owners.insert({ id: ownerId, photo_id: attachment.id }) + if (command === `reject-save`) + throw new Error(`injected hook failure`) + }, + }) + if (command === `reject-save`) { + await expect(saving).rejects.toThrow(`injected hook failure`) + } else { + const saved = await saving + uri = saved.local_uri ?? undefined + model.intent = `present` + } + break + } + case `duplicate-warm`: + case `duplicate-cold`: { + const collection = + command === `duplicate-cold` + ? createCollection( + powerSyncCollectionOptions({ + database: db, + table: schema.props.attachments, + }), + ) + : attachments + const duplicateQueue = new TanStackDBAttachmentQueue({ + db, + attachmentsCollection: collection, + localStorage: local, + remoteStorage: transport, + watchAttachments: () => {}, + }) + try { + if (command === `duplicate-cold`) + expect(collection.get(attachmentId)).toBeUndefined() + await expect( + duplicateQueue.save({ + id: attachmentId, + fileExtension: `bin`, + data: new Uint8Array(model.bytes.map((byte) => byte ^ 255)) + .buffer, + updateHook: () => + owners.update(ownerId, (owner) => { + owner.photo_id = null + }), + }), + ).rejects.toThrow() + } finally { + await duplicateQueue.stopSync() + if (collection !== attachments) await collection.cleanup() + } + break + } + case `delete`: + case `reject-delete`: { + const deleting = queue.delete({ + id: attachmentId, + updateHook: () => { + owners.update(ownerId, (owner) => { + owner.photo_id = null + }) + if (command === `reject-delete`) + throw new Error(`injected hook failure`) + }, + }) + if (command === `reject-delete`) { + await expect(deleting).rejects.toThrow(`injected hook failure`) + } else { + await deleting + model.intent = `deleted` + } + break + } + case `start-upload`: + if (syncing) throw new Error(`upload already in flight`) + uploadGate = { entered: pDefer(), outcome: pDefer() } + syncing = queue.syncStorage() + await vi.waitFor(() => expect(io.uploads).toBeGreaterThan(0), { + timeout: 3000, + interval: 10, + }) + await uploadGate.entered.promise + break + case `finish-upload`: + await finishUpload(true) + break + case `fail-upload`: + await finishUpload(false) + break + case `drain`: { + if (syncing) throw new Error(`finish the held upload before draining`) + // No more injected failures: two passes suffice for these single-ID + // histories. A third tick must be idle, not merely leave SQL detached. + await queue.syncStorage() + await queue.syncStorage() + const settledIO = { ...io } + await queue.syncStorage() + expect(io).toEqual(settledIO) + break + } + } + } + + return { + run, + assertObserved, + async dispose() { + if (uploadGate) uploadGate.outcome.resolve(true) + await syncing + await queue.stopSync() + await Promise.all([attachments.cleanup(), owners.cleanup()]) + await db.disconnectAndClear() + await db.close() + await local.clear() + await rm(directory, { recursive: true }) + }, + } +} + +export async function runHistory(history: History) { + const fixture = await setupOracle() + const model: Model = { intent: `absent`, bytes: [...history.bytes] } + try { + for (const [index, command] of history.commands.entries()) { + const label = `${history.name}: step ${index + 1}/${history.commands.length} ${command}; ${JSON.stringify(history)}` + await fixture.run(command, model) + await fixture.assertObserved(model, label, command === `drain`) + } + } finally { + await fixture.dispose() + } +} + +export const corpus: Array> = [ + { + name: `warm duplicate preserves bytes and reference`, + commands: [`save`, `duplicate-warm`, `drain`], + }, + { + name: `cold duplicate preserves bytes and reference`, + commands: [`save`, `duplicate-cold`, `drain`], + }, + { + name: `delete before upload removes local data`, + commands: [`save`, `delete`, `drain`], + }, + { + name: `delete during successful upload removes remote data`, + commands: [`save`, `start-upload`, `delete`, `finish-upload`, `drain`], + }, + { + name: `delete after upload removes remote data`, + commands: [`save`, `start-upload`, `finish-upload`, `delete`, `drain`], + }, + { + name: `delete during failed upload does not retry obsolete intent`, + commands: [`save`, `start-upload`, `delete`, `fail-upload`, `drain`], + }, + { + name: `upload failure retries accepted bytes`, + commands: [`save`, `start-upload`, `fail-upload`, `drain`], + }, + { + name: `delete after upload failure removes local data`, + commands: [`save`, `start-upload`, `fail-upload`, `delete`, `drain`], + }, + { + name: `save hook failure rolls back both rows and local data`, + commands: [`reject-save`, `save`, `drain`], + }, + { + name: `delete hook failure preserves bytes and reference`, + commands: [`save`, `reject-delete`, `drain`], + }, + { + name: `repeated delete stays deleted`, + commands: [`save`, `delete`, `delete`, `drain`], + }, +] + +// The ordinary suite excludes only the named SDK completion race; the opt-in +// repro suite runs that same law and harness without an expected-failure waiver. +export const sdkCompletionCorpus = corpus.filter( + ({ commands }) => + commands.indexOf(`delete`) > commands.indexOf(`start-upload`) && + commands.indexOf(`delete`) < + Math.max( + commands.indexOf(`finish-upload`), + commands.indexOf(`fail-upload`), + ), +) +export const supportedCorpus = corpus.filter( + (history) => !sdkCompletionCorpus.includes(history), +) + +export function historyArbitrary(includeInFlightDelete: boolean) { + return fc + .record({ + bytes: fc.array(fc.integer({ min: 0, max: 255 }), { + minLength: 1, + maxLength: 16, + }), + duplicate: fc.constantFrom(`none`, `warm`, `cold`), + deletion: includeInFlightDelete + ? fc.constantFrom(`never`, `before`, `during`, `after`) + : fc.constantFrom(`never`, `before`, `after`), + uploadSucceeds: fc.boolean(), + rejectDelete: fc.boolean(), + }) + .map( + ({ + bytes, + duplicate, + deletion, + uploadSucceeds, + rejectDelete, + }): History => { + const commands: Array = [`save`] + if (duplicate === `warm`) commands.push(`duplicate-warm`) + if (duplicate === `cold`) commands.push(`duplicate-cold`) + if (rejectDelete) commands.push(`reject-delete`) + if (deletion === `before`) { + commands.push(`delete`) + } else { + commands.push(`start-upload`) + if (deletion === `during`) commands.push(`delete`) + commands.push(uploadSucceeds ? `finish-upload` : `fail-upload`) + if (deletion === `after`) commands.push(`delete`) + } + commands.push(`drain`) + return { name: `generated`, commands, bytes } + }, + ) +} diff --git a/packages/powersync-db-collection/tests/attachments-lifecycle-oracle.test.ts b/packages/powersync-db-collection/tests/attachments-lifecycle-oracle.test.ts new file mode 100644 index 000000000..2b5e50d44 --- /dev/null +++ b/packages/powersync-db-collection/tests/attachments-lifecycle-oracle.test.ts @@ -0,0 +1,37 @@ +import { fc } from '@fast-check/vitest' +import { describe, it } from 'vitest' +import { + historyArbitrary, + runHistory, + supportedCorpus, +} from './attachments-lifecycle-fixture' +import { TEST_DATABASE_IMPLEMENTATION } from './test-db-implementation' + +const describePowerSync = TEST_DATABASE_IMPLEMENTATION + ? describe + : describe.skip + +describePowerSync(`attachment lifecycle intent oracle`, () => { + it.each(supportedCorpus)(`$name`, async (history) => { + await runHistory({ ...history, bytes: [0, 7, 128, 255] }) + }) + + it.each([`fixed`, `random`] as const)( + `preserves intent across %s command histories`, + async (campaign) => { + const replaySeed = process.env.POWERSYNC_ATTACHMENT_ORACLE_SEED + await fc.assert(fc.asyncProperty(historyArbitrary(false), runHistory), { + ...(replaySeed + ? { seed: Number(replaySeed) } + : campaign === `fixed` + ? { seed: 1616 } + : {}), + numRuns: 12, + ...(process.env.POWERSYNC_ATTACHMENT_ORACLE_PATH + ? { path: process.env.POWERSYNC_ATTACHMENT_ORACLE_PATH } + : {}), + }) + }, + 30000, + ) +}) diff --git a/packages/powersync-db-collection/tests/attachments-native-sdk.repro.ts b/packages/powersync-db-collection/tests/attachments-native-sdk.repro.ts new file mode 100644 index 000000000..81920c1b8 --- /dev/null +++ b/packages/powersync-db-collection/tests/attachments-native-sdk.repro.ts @@ -0,0 +1,85 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { + AttachmentQueue, + AttachmentState, + AttachmentTable, + Schema, +} from '@powersync/common' +import { NodeFileSystemAdapter, PowerSyncDatabase } from '@powersync/node' +import pDefer from 'p-defer' +import { describe, expect, it } from 'vitest' +import { TEST_DATABASE_IMPLEMENTATION } from './test-db-implementation' + +const describePowerSync = TEST_DATABASE_IMPLEMENTATION + ? describe + : describe.skip + +// No TanStack collection, subclass, transaction, or watcher participates in this +// control. The native SDK alone loses deletion intent in its completion write. +describePowerSync(`native SDK completion contract`, () => { + it.each([true, false])( + `keeps deletion queued after upload success=%s`, + async (succeeds) => { + const directory = await mkdtemp(join(tmpdir(), `ps-native-completion-`)) + const db = new PowerSyncDatabase({ + database: { + dbFilename: `attachments.sqlite`, + dbLocation: directory, + implementation: TEST_DATABASE_IMPLEMENTATION, + }, + schema: new Schema({ attachments: new AttachmentTable() }), + }) + await db.disconnectAndClear() + const localStorage = new NodeFileSystemAdapter(join(directory, `files`)) + await localStorage.initialize() + const entered = pDefer() + const release = pDefer() + const queue = new AttachmentQueue({ + db, + localStorage, + remoteStorage: { + async uploadFile() { + entered.resolve() + await release.promise + if (!succeeds) throw new Error(`injected upload failure`) + }, + async deleteFile() {}, + downloadFile() { + throw new Error(`unexpected download`) + }, + }, + watchAttachments: () => {}, + }) + let syncing: Promise | undefined + try { + const record = await queue.saveFile({ + data: new Uint8Array([0, 7, 128, 255]).buffer, + fileExtension: `bin`, + }) + syncing = queue.syncStorage() + await entered.promise + await queue.deleteFile({ id: record.id }) + const readState = () => + db.get(`SELECT state FROM attachments WHERE id = ?`, [record.id]) + expect(await readState()).toEqual({ + state: AttachmentState.QUEUED_DELETE, + }) + release.resolve() + await syncing + expect(await readState()).toEqual({ + state: AttachmentState.QUEUED_DELETE, + }) + } finally { + release.resolve() + await syncing + await queue.stopSync() + await db.disconnectAndClear() + await db.close() + await localStorage.clear() + await rm(directory, { recursive: true }) + } + }, + ) +}) diff --git a/packages/powersync-db-collection/tests/attachments-sdk-completion.repro.ts b/packages/powersync-db-collection/tests/attachments-sdk-completion.repro.ts new file mode 100644 index 000000000..0026ab943 --- /dev/null +++ b/packages/powersync-db-collection/tests/attachments-sdk-completion.repro.ts @@ -0,0 +1,30 @@ +import { fc } from '@fast-check/vitest' +import { describe, it } from 'vitest' +import { + historyArbitrary, + runHistory, + sdkCompletionCorpus, +} from './attachments-lifecycle-fixture' +import { TEST_DATABASE_IMPLEMENTATION } from './test-db-implementation' + +const describePowerSync = TEST_DATABASE_IMPLEMENTATION + ? describe + : describe.skip + +// Desired contract, deliberately not test.fails: run separately until the SDK +// preserves newer deletion intent in both successful and failed upload writes. +describePowerSync(`upstream attachment completion contract`, () => { + it.each(sdkCompletionCorpus)(`$name`, async (history) => { + await runHistory({ ...history, bytes: [0, 7, 128, 255] }) + }) + + it(`preserves deletion across all generated completion schedules`, async () => { + await fc.assert(fc.asyncProperty(historyArbitrary(true), runHistory), { + seed: Number(process.env.POWERSYNC_ATTACHMENT_ORACLE_SEED ?? 1616), + numRuns: 12, + ...(process.env.POWERSYNC_ATTACHMENT_ORACLE_PATH + ? { path: process.env.POWERSYNC_ATTACHMENT_ORACLE_PATH } + : {}), + }) + }, 30000) +}) diff --git a/packages/powersync-db-collection/tests/attachments.test.ts b/packages/powersync-db-collection/tests/attachments.test.ts index 193f8fa0c..1cafe169f 100644 --- a/packages/powersync-db-collection/tests/attachments.test.ts +++ b/packages/powersync-db-collection/tests/attachments.test.ts @@ -1,6 +1,7 @@ import { randomUUID } from 'node:crypto' import { tmpdir } from 'node:os' import { join } from 'node:path' +import pDefer from 'p-defer' import { AttachmentState, AttachmentTable, @@ -56,7 +57,7 @@ const describePowerSync = TEST_DATABASE_IMPLEMENTATION : describe.skip describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { - async function setup() { + async function setup(syncMode: `eager` | `on-demand` = `eager`) { const db = new PowerSyncDatabase({ database: { dbFilename: `attachments-test-${randomUUID()}.sqlite`, @@ -91,6 +92,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { powerSyncCollectionOptions({ database: db, table: APP_SCHEMA.props.attachments, + syncMode, }), ) const usersCollection = createCollection( @@ -212,6 +214,292 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { } describe(`save`, () => { + it(`serializes successive watched snapshots through the SDK context`, async () => { + const fixture = await setup() + let update!: Parameters[0] + const queue = fixture.createQueue({ + archivedCacheLimit: 100, + watchAttachments: (callback) => { + update = callback + }, + }) + const record = await queue.save({ + data: createMockJpegBuffer(), + fileExtension: `jpg`, + }) + await queue.syncStorage() + await queue.startSync() + const entered = pDefer() + const release = pDefer() + const held = queue.withAttachmentContext(async () => { + entered.resolve() + await release.promise + }) + await entered.promise + const completed: Array = [] + const first = update([]).then(() => { + completed.push(1) + }) + const second = update([{ id: record.id, fileExtension: `jpg` }]).then( + () => { + completed.push(2) + }, + ) + try { + await Promise.resolve() + expect(completed).toEqual([]) + release.resolve() + await Promise.all([held, first, second]) + expect(completed).toEqual([1, 2]) + expect( + await fixture.db.get(`SELECT state FROM attachments WHERE id = ?`, [ + record.id, + ]), + ).toEqual({ state: AttachmentState.SYNCED }) + } finally { + release.resolve() + await Promise.all([held, first, second]) + } + }) + + it(`preserves the winner file across two queue instances sharing storage`, async () => { + const fixture = await setup() + const first = fixture.createQueue() + const second = fixture.createQueue() + const entered = pDefer() + const release = pDefer() + const saveFile = fixture.localStorage.saveFile.bind(fixture.localStorage) + const write = vi + .spyOn(fixture.localStorage, `saveFile`) + .mockImplementation(async (...args) => { + const size = await saveFile(...args) + if (write.mock.calls.length === 1) { + entered.resolve() + await release.promise + } + return size + }) + const saving = first.save({ + id: `shared-id`, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + }) + try { + await entered.promise + await expect( + second.save({ + id: `shared-id`, + data: new Uint8Array([7, 8, 9]).buffer, + fileExtension: `jpg`, + }), + ).rejects.toThrow(/already/) + expect(write).toHaveBeenCalledTimes(1) + } finally { + release.resolve() + await saving + } + const winner = await saving + expect( + new Uint8Array(await fixture.localStorage.readFile(winner.local_uri!)), + ).toEqual(new Uint8Array(createMockJpegBuffer())) + }) + + it.each([`ready`, `cold`, `on-demand`] as const)( + `preserves an existing file when reused through a %s collection`, + async (phase) => { + const fixture = await setup() + const original = await fixture.createQueue().save({ + id: `reused-after-reopen`, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + }) + const collection = + phase === `ready` + ? fixture.attachmentsCollection + : createCollection( + powerSyncCollectionOptions({ + database: fixture.db, + table: APP_SCHEMA.props.attachments, + syncMode: phase === `on-demand` ? `on-demand` : `eager`, + }), + ) + onTestFinished(() => collection.cleanup()) + const queue = new TanStackDBAttachmentQueue({ + db: fixture.db, + attachmentsCollection: collection, + localStorage: fixture.localStorage, + remoteStorage: fixture.remoteStorage, + watchAttachments: () => {}, + }) + onTestFinished(() => queue.stopSync()) + if (phase !== `ready`) + expect(collection.get(original.id)).toBeUndefined() + await expect( + queue.save({ + id: original.id, + data: new Uint8Array([7, 8, 9]).buffer, + fileExtension: `jpg`, + }), + ).rejects.toThrow() + expect( + await fixture.db.getOptional( + `SELECT id FROM attachments WHERE id = ?`, + [original.id], + ), + ).toEqual({ id: original.id }) + expect(await fixture.localStorage.fileExists(original.local_uri!)).toBe( + true, + ) + expect( + new Uint8Array( + await fixture.localStorage.readFile(original.local_uri!), + ), + ).toEqual(new Uint8Array(createMockJpegBuffer())) + }, + ) + + it(`cleans a partial write and releases its ID for retry`, async () => { + const fixture = await setup() + const write = fixture.localStorage.saveFile.bind(fixture.localStorage) + const savedPaths: Array = [] + vi.spyOn(fixture.localStorage, `saveFile`).mockImplementationOnce( + async (uri, data) => { + savedPaths.push(uri) + await write(uri, data) + throw new Error(`partial write failure`) + }, + ) + const options = { + id: `partial`, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + } + await expect(fixture.createQueue().save(options)).rejects.toThrow( + `partial write failure`, + ) + expect(savedPaths).toHaveLength(1) + expect(await fixture.localStorage.fileExists(savedPaths[0]!)).toBe(false) + expect( + await fixture.db.getOptional( + `SELECT id FROM attachments WHERE id = ?`, + [options.id], + ), + ).toBeNull() + const saved = await fixture.createQueue().save(options) + expect(await fixture.localStorage.fileExists(saved.local_uri!)).toBe(true) + }) + + it(`allows another queue to save a distinct ID while a write is held`, async () => { + const fixture = await setup() + const entered = pDefer() + const release = pDefer() + const write = fixture.localStorage.saveFile.bind(fixture.localStorage) + vi.spyOn(fixture.localStorage, `saveFile`).mockImplementationOnce( + async (...args) => { + const size = await write(...args) + entered.resolve() + await release.promise + return size + }, + ) + const saving = fixture + .createQueue() + .save({ + id: `held`, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + }) + try { + await entered.promise + const other = await fixture + .createQueue() + .save({ + id: `other`, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + }) + expect(await fixture.localStorage.fileExists(other.local_uri!)).toBe( + true, + ) + } finally { + release.resolve() + await saving + } + }) + + it.each([`before`, `after`] as const)( + `preserves delete intent %s an SDK upload`, + async (timing) => { + const fixture = await setup() + const queue = fixture.createQueue() + const uploaded = pDefer() + const release = pDefer() + const remoteFiles = new Set() + fixture.uploadFile.mockImplementation(async (_, attachment) => { + uploaded.resolve() + await release.promise + remoteFiles.add(attachment.id) + }) + fixture.deleteFile.mockImplementation((attachment) => { + remoteFiles.delete(attachment.id) + return Promise.resolve() + }) + const userId = randomUUID() + const record = await queue.save({ + data: createMockJpegBuffer(), + fileExtension: `jpg`, + updateHook: (attachment) => { + fixture.usersCollection.insert({ + id: userId, + name: `owner`, + email: null, + photo_id: attachment.id, + }) + }, + }) + let sync: Promise | undefined + try { + if (timing !== `before`) { + sync = queue.syncStorage() + await uploaded.promise + release.resolve() + await sync + } + await queue.delete({ + id: record.id, + updateHook: () => { + fixture.usersCollection.update(userId, (row) => { + row.photo_id = null + }) + }, + }) + expect( + await fixture.db.get(`SELECT state FROM attachments WHERE id = ?`, [ + record.id, + ]), + ).toEqual({ state: AttachmentState.QUEUED_DELETE }) + expect( + await fixture.db.get(`SELECT photo_id FROM users WHERE id = ?`, [ + userId, + ]), + ).toEqual({ photo_id: null }) + release.resolve() + await sync + // Complete two real SDK passes, not a timeout waiting for a mock state. + await queue.syncStorage() + await queue.syncStorage() + expect(fixture.deleteFile).toHaveBeenCalledTimes(1) + expect(remoteFiles.has(record.id)).toBe(false) + expect(await fixture.localStorage.fileExists(record.local_uri!)).toBe( + false, + ) + } finally { + release.resolve() + await sync + } + }, + ) + it(`writes the local file and inserts a QUEUED_UPLOAD row into the collection`, async () => { const { createQueue, attachmentsCollection, localStorage } = await setup() const queue = createQueue() @@ -376,7 +664,9 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { expect(fulfilled).toHaveLength(1) expect(rejected).toHaveLength(1) expect(rejected[0]!.reason).toEqual( - expect.objectContaining({ message: expect.stringMatching(/exists/) }), + expect.objectContaining({ + message: expect.stringMatching(/exists|being saved/), + }), ) const winner = fulfilled[0]!.value @@ -398,9 +688,8 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { } = await setup() const queue = createQueue() - // A caller-supplied id lets us derive the local uri without a returned record. const id = randomUUID() - const localUri = localStorage.getLocalUri(`${id}.jpg`) + let localUri: string | undefined await expect( queue.save({ @@ -408,6 +697,7 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { data: createMockJpegBuffer(), fileExtension: `jpg`, updateHook: (attachment) => { + localUri = attachment.local_uri! usersCollection.insert({ id: randomUUID(), name: `steven`, @@ -420,7 +710,8 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { ).rejects.toThrow(/updateHook failed/) // The file is written before the transaction opens, so it must be cleaned up. - expect(await localStorage.fileExists(localUri)).toBe(false) + expect(localUri).toBeDefined() + expect(await localStorage.fileExists(localUri!)).toBe(false) // Neither the attachment nor the hook's own mutation may survive the failure. expect(attachmentsCollection.get(id)).toBeUndefined() @@ -429,6 +720,110 @@ describePowerSync(`PowerSync AttachmentQueue (TanStackDB)`, () => { }) describe(`delete file`, () => { + it.each([`eager`, `on-demand`] as const)( + `can retry a failed save and later delete in %s mode`, + async (syncMode) => { + const fixture = await setup(syncMode) + const first = fixture.createQueue() + const options = { + id: `retry`, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + } + await expect( + first.save({ + ...options, + updateHook: () => { + throw new Error(`hook failure`) + }, + }), + ).rejects.toThrow(`hook failure`) + const second = fixture.createQueue() + const saved = await second.save(options) + expect( + new Uint8Array(await fixture.localStorage.readFile(saved.local_uri!)), + ).toEqual(new Uint8Array(createMockJpegBuffer())) + await first.delete({ id: saved.id }) + expect( + await fixture.db.get(`SELECT state FROM attachments WHERE id = ?`, [ + saved.id, + ]), + ).toEqual({ state: AttachmentState.QUEUED_DELETE }) + }, + ) + + it(`finds a queued file after its storage root moves`, async () => { + const fixture = await setup() + const original = await fixture.createQueue().save({ + id: `moving`, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + }) + const moved = new NodeFileSystemAdapter( + join(tmpdir(), `ps-moved-${randomUUID()}`), + ) + await moved.initialize() + onTestFinished(() => moved.clear()) + // Move the bytes without changing SQLite, as a changed app directory does. + const movedUri = moved.getLocalUri(original.local_uri!.split(`/`).at(-1)!) + await moved.saveFile( + movedUri, + await fixture.localStorage.readFile(original.local_uri!), + ) + await fixture.localStorage.deleteFile(original.local_uri!) + const queue = new TanStackDBAttachmentQueue({ + db: fixture.db, + attachmentsCollection: fixture.attachmentsCollection, + localStorage: moved, + remoteStorage: fixture.remoteStorage, + watchAttachments: () => {}, + }) + onTestFinished(() => queue.stopSync()) + await queue.startSync() + await vi.waitFor(() => + expect(fixture.uploadFile).toHaveBeenCalledTimes(1), + ) + expect(fixture.uploadFile.mock.calls[0]![1].localUri).toBe(movedUri) + expect(new Uint8Array(await moved.readFile(movedUri))).toEqual( + new Uint8Array(createMockJpegBuffer()), + ) + }) + + it.each([`eager`, `on-demand`] as const)( + `loads an uncached attachment before deleting in %s mode`, + async (syncMode) => { + const fixture = await setup() + const original = await fixture.createQueue().save({ + id: `uncached`, + data: createMockJpegBuffer(), + fileExtension: `jpg`, + }) + const collection = createCollection( + powerSyncCollectionOptions({ + database: fixture.db, + table: APP_SCHEMA.props.attachments, + syncMode, + }), + ) + onTestFinished(() => collection.cleanup()) + const queue = new TanStackDBAttachmentQueue({ + db: fixture.db, + attachmentsCollection: collection, + localStorage: fixture.localStorage, + remoteStorage: fixture.remoteStorage, + watchAttachments: () => {}, + }) + onTestFinished(() => queue.stopSync()) + expect(collection.get(original.id)).toBeUndefined() + await queue.delete({ id: original.id }) + expect( + await fixture.db.get(`SELECT state FROM attachments WHERE id = ?`, [ + original.id, + ]), + ).toEqual({ state: AttachmentState.QUEUED_DELETE }) + }, + ) + it(`queues an existing attachment for deletion and removes the local file`, async () => { const { createQueue, diff --git a/packages/powersync-db-collection/tests/upstream.config.ts b/packages/powersync-db-collection/tests/upstream.config.ts new file mode 100644 index 000000000..2dd49b263 --- /dev/null +++ b/packages/powersync-db-collection/tests/upstream.config.ts @@ -0,0 +1,12 @@ +import { defineConfig } from 'vitest/config' + +export default defineConfig({ + test: { + name: `powersync-upstream-contract-repros`, + dir: `./tests`, + include: [`**/*.repro.ts`], + environment: `node`, + coverage: { enabled: false }, + typecheck: { enabled: false }, + }, +})