fix: preserve attachment files across startup and concurrent saves

.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.

.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.

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,
     }))
 )
-```
+```

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",

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<AbstractPowerSyncDatabase, Set<string>>()
+
 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<void> {
+    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<void> {
-    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<T>(
+    id: string,
+    operation: () => Promise<T>,
+  ): Promise<T> {
+    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()
+    }
   }
 
   /**

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.