refactor(sync): incremental Pierre persistence with per-record files (tldraw#8013)

In order to enable granular git-level diffs in Pierre history, this PR
rewrites Pierre persistence to store each tldraw record as an individual
file and use the `TLSyncStorage` interface for incremental updates.

**Pierre storage format**: Instead of a single `snapshot.json` blob,
each commit now contains `meta.json` (documentClock, schema only) and
individual `records/{id}.json` files. Each record file stores only the
record state (no lastChangedClock). Tombstone metadata is not stored in
Pierre; the file-per-record model represents deletes by removing files.
This gives meaningful per-record diffs in Pierre's commit history.

**Incremental persistence**: Uses
`storage.getChangesSince(pierreDocClock)` to compute the diff since the
last Pierre commit. Only changed records are written per commit. For
empty repos, `documentClock=-1` triggers returning all records. On
`wipeAll`, stale record files in Pierre are cleaned up via
`repo.listFiles()`. Optimistic concurrency via `expectedHeadSha` with
retry on CAS conflict.

**getChangesSince(wipeAll)**: When `wipeAll` is true (e.g. empty repo,
`sinceClock -1`), storage was still including tombstones in
`diff.deletes`, which caused the first Pierre commit to call
`deletePath` on non-existent files and could duplicate deletes with the
cleanup loop. InMemorySyncStorage and SQLiteSyncStorage now omit
`diff.deletes` when `wipeAll`; the durable object applies `diff.deletes`
only when `headSha && !wipeAll`.

**Restore fix**: Pierre restore no longer calls `repo.restoreCommit()`
(which failed when restoring to HEAD). Instead it reconstructs the
snapshot from the Pierre archive via `reconstructSnapshotFromPierre` and
loads it into storage with `loadSnapshotIntoStorage`, so the next
persist cycle writes the restored state as a new forward commit.

**Snapshot reconstruction**: New `pierreSnapshot.ts` uses `modern-tar`'s
`createTarDecoder()` to stream the gzip-compressed tar archive from
Pierre. The pipeline is `getArchiveStream` → DecompressionStream('gzip')
→ createTarDecoder(); entries are consumed with a reader loop. Record
files are parsed as state only; reconstructed documents use
`lastChangedClock: 0` (loadSnapshotIntoStorage only uses state). History
snapshot and restore both call `reconstructSnapshotFromPierre` and
return or load the resulting `RoomSnapshot`.

**Other**: Added missing `assert` import in `TLSyncStorage.ts`.
`syncPierreState` uses `listCommits({ limit: 1 })` and
`getFileStream('meta.json')` to sync HEAD and documentClock. Pierre repo
IDs changed from `.../snapshots/...` to `.../files/...`. Upgraded
`@pierre/storage` to `^1.1.0` and added `modern-tar`. Client history
snapshot page uses `TlaFileError` in the error boundary and drops the
unused error prop from the main component.

### Change type

- [x] `improvement`

### Test plan

1. Open a tldraw file with Pierre enabled
2. Make edits and verify Pierre commits contain individual
`records/*.json` entries (state only)
3. View Pierre history — verify snapshot loading works
4. Restore from Pierre history — verify the canvas updates and a new
Pierre commit is created
5. Verify incremental commits only touch changed records
6. New file (empty Pierre repo): first persist should succeed without
deletePath errors

- [ ] Unit tests
- [ ] End to end tests

### Release notes

- Internal improvement to Pierre version history storage format for
better granularity.

---------

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: ds300

apps/dotcom/client/src/tla/pages/file-pierre-history-snapshot.tsx

@@ -16,7 +16,7 @@ export function ErrorBoundary() {
 	useEffect(() => {
 		captureException(error)
 	}, [error])
-	return <Component error={error} />
+	return <TlaFileError error={error} />
 }
 
 const { loader, useData } = defineLoader(async (args) => {
@@ -36,7 +36,7 @@ const { loader, useData } = defineLoader(async (args) => {
 
 export { loader }
 
-export function Component({ error: _error }: { error?: unknown }) {
+export function Component() {
 	const userId = useMaybeApp()?.userId
 
 	const result = useData()
@@ -54,7 +54,7 @@ export function Component({ error: _error }: { error?: unknown }) {
 		} as TLStoreSnapshot
 	}, [result])
 
-	const error = _error || !result || !snapshot
+	const error = !result || !snapshot
 
 	useEffect(() => {
 		if (error && userId) {

apps/dotcom/sync-worker/package.json

@@ -23,7 +23,7 @@
 	},
 	"dependencies": {
 		"@clerk/backend": "^1.23.7",
-		"@pierre/storage": "^0.1.1",
+		"@pierre/storage": "^1.1.0",
 		"@rocicorp/zero": "0.25.9",
 		"@supabase/auth-helpers-remix": "^0.2.6",
 		"@supabase/supabase-js": "^2.48.1",
@@ -39,6 +39,7 @@
 		"jose": "^5.9.6",
 		"kysely": "^0.27.5",
 		"lodash.throttle": "^4.1.1",
+		"modern-tar": "^0.7.5",
 		"pg": "^8.13.1",
 		"pg-logical-replication": "^2.0.7",
 		"react": "^19.2.1",

apps/dotcom/sync-worker/src/TLDrawDurableObject.ts

@@ -1,7 +1,7 @@
 /// <reference no-default-lib="true"/>
 /// <reference types="@cloudflare/workers-types" />
 
-import { RefUpdateError } from '@pierre/storage'
+import { ApiError, RefUpdateError, type Repo } from '@pierre/storage'
 import { SupabaseClient } from '@supabase/supabase-js'
 import {
 	DB,
@@ -53,6 +53,7 @@ import { EventData, writeDataPoint } from './utils/analytics'
 import { createPierreClient } from './utils/createPierreClient'
 import { createSupabaseClient } from './utils/createSupabaseClient'
 import { getRoomDurableObject } from './utils/durableObjects'
+import { reconstructSnapshotFromPierre } from './utils/pierreSnapshot'
 import { isRateLimited } from './utils/rateLimit'
 import { getSlug } from './utils/roomOpenMode'
 import { throttle } from './utils/throttle'
@@ -211,6 +212,7 @@ export class TLFileDurableObject extends DurableObject {
 	// For persistence
 	supabaseClient: SupabaseClient | void
 	pierreClient: ReturnType<typeof createPierreClient>
+	pierreState: PierreState | null = null
 
 	// For analytics
 	measure: Analytics | undefined
@@ -371,16 +373,9 @@ export class TLFileDurableObject extends DurableObject {
 				if (!repo) {
 					return new Response('Pierre not available', { status: 503 })
 				}
-				const fileStream = await repo.getFileStream({
-					path: SNAPSHOT_FILE_NAME,
-					ref: commitHash,
-				})
-				dataText = await fileStream.text()
-				await repo.restoreCommit({
-					targetCommitSha: commitHash,
-					targetBranch: 'main',
-					author: PIERRE_AUTHOR,
-				})
+				const snapshot = await reconstructSnapshotFromPierre(repo, commitHash)
+				dataText = JSON.stringify(snapshot)
+				this.pierreState = null
 			} else {
 				const timestamp = ((await req.json()) as any).timestamp
 				if (!timestamp) {
@@ -973,6 +968,7 @@ export class TLFileDurableObject extends DurableObject {
 
 						const key = getR2KeyForRoom({ slug: slug, isApp: this.documentInfo.isApp })
 						await this._uploadSnapshotToR2(snapshot, key)
+						await this.persistToPierre(storage, snapshot)
 
 						this.logEvent({ type: 'persist_success', attempts: attempt })
 						this._lastPersistedClock = snapshot.documentClock
@@ -1023,7 +1019,6 @@ export class TLFileDurableObject extends DurableObject {
 		// Then upload to version cache
 		const versionKey = `${key}/${new Date().toISOString()}`
 		await this._uploadSnapshotToBucket(this.r2.versionCache, snapshot, versionKey)
-		await this.persistToPierre(versionKey)
 	}
 
 	private async _uploadSnapshotToBucket(bucket: R2Bucket, snapshot: RoomSnapshot, key: string) {
@@ -1120,44 +1115,139 @@ export class TLFileDurableObject extends DurableObject {
 		) {
 			return null
 		}
-		const repoId = `${this.env.TLDRAW_ENV}/snapshots/${this.documentInfo.slug}`
+		const repoId = `${this.env.TLDRAW_ENV}/files/${this.documentInfo.slug}`
 		return (
 			(await this.pierreClient.findOne({ id: repoId })) ??
 			(await this.pierreClient.createRepo({ id: repoId }))
 		)
 	}
 
-	private async persistToPierre(key: string) {
+	/**
+	 * Sync local Pierre tracking state from the remote repo. Fetches HEAD sha
+	 * and meta.json (for documentClock). For empty repos, sets documentClock
+	 * to -1 so getChangesSince returns all records.
+	 */
+	private async syncPierreState(repo: Repo) {
+		let headCommit: { sha: string } | undefined
+		try {
+			const { commits } = await repo.listCommits({ limit: 1 })
+			headCommit = commits[0]
+		} catch (error) {
+			if (error instanceof ApiError && error.status === 404) {
+				this.pierreState = { headSha: undefined, documentClock: -1 }
+				return
+			}
+			throw error
+		}
+
+		if (!headCommit) {
+			this.pierreState = { headSha: undefined, documentClock: -1 }
+			return
+		}
+
+		const metaResp = await repo.getFileStream({ path: 'meta.json', ref: headCommit.sha })
+		const meta = (await metaResp.json()) as PierreMeta
+
+		this.pierreState = {
+			headSha: headCommit.sha,
+			documentClock: meta.documentClock ?? 0,
+		}
+	}
+
+	private async persistToPierre(storage: TLSyncStorage<TLRecord>, snapshot: RoomSnapshot) {
 		try {
 			const repo = await this.getPierreRepo()
 			if (!repo) return
 
-			// Get the snapshot from R2 to create a readable stream
-			const r2Object = await this.r2.versionCache.get(key)
-			if (!r2Object) {
-				console.warn('Failed to get R2 object for Pierre upload:', key)
-				return
-			}
+			const MAX_CAS_RETRIES = 3
+			for (let attempt = 0; attempt < MAX_CAS_RETRIES; attempt++) {
+				if (!this.pierreState) {
+					await this.syncPierreState(repo)
+				}
+
+				const { headSha, documentClock: pierreDocClock } = this.pierreState!
+
+				const { result: changes, documentClock } = storage.transaction((txn) =>
+					txn.getChangesSince(pierreDocClock)
+				)
+
+				if (!changes) return
+
+				const { diff } = changes
+				const hasPuts = Object.keys(diff.puts).length > 0
+				const hasDeletes = diff.deletes.length > 0
 
-			// Create commit with the snapshot
-			const timestamp = new Date().toISOString()
-			await repo
-				.createCommit({
+				if (!hasPuts && !hasDeletes && pierreDocClock === documentClock) {
+					return
+				}
+
+				const timestamp = new Date().toISOString()
+				const commitBuilder = repo.createCommit({
 					targetBranch: 'main',
 					commitMessage: `Snapshot at ${timestamp}`,
 					author: PIERRE_AUTHOR,
+					expectedHeadSha: headSha,
 				})
-				.addFile(SNAPSHOT_FILE_NAME, r2Object.body)
-				.send()
-				.catch((e) => {
-					// ignore no changes to commit errors
-					if (e instanceof RefUpdateError && e.message.match('no changes to commit')) {
-						return
+
+				const meta: PierreMeta = {
+					documentClock,
+					schema: snapshot.schema,
+				}
+				commitBuilder.addFileFromString('meta.json', JSON.stringify(meta))
+
+				for (const [id, put] of Object.entries(diff.puts)) {
+					const state = Array.isArray(put) ? put[1] : put
+					commitBuilder.addFileFromString(`records/${id}.json`, JSON.stringify(state))
+				}
+
+				// Only apply diff.deletes when we have a parent commit and we're not in wipeAll.
+				// - Empty repo (no headSha): those paths don't exist in Pierre; deletePath would fail.
+				// - wipeAll with existing repo: the cleanup loop below already deletes any file not in
+				//   putIds, so applying diff.deletes here would duplicate deletePath for the same file.
+				if (headSha && !changes.wipeAll) {
+					for (const id of diff.deletes) {
+						commitBuilder.deletePath(`records/${id}.json`)
 					}
-					throw e
-				})
+				}
+
+				// On wipeAll with an existing repo, pruned tombstones may not appear in diff.deletes,
+				// so scan Pierre for stale record files and remove them.
+				if (changes.wipeAll && headSha) {
+					const putIds = new Set(Object.keys(diff.puts))
+					const { paths } = await repo.listFiles({ ref: headSha })
+					for (const path of paths) {
+						if (!path.startsWith('records/')) continue
+						const id = path.slice('records/'.length, -'.json'.length)
+						if (!putIds.has(id)) {
+							commitBuilder.deletePath(path)
+						}
+					}
+				}
+
+				try {
+					const result = await commitBuilder.send().catch((e) => {
+						if (e instanceof RefUpdateError && e.message.match('no changes to commit')) {
+							return null
+						}
+						throw e
+					})
+
+					this.pierreState = {
+						headSha: result ? result.refUpdate.newSha : headSha,
+						documentClock,
+					}
+					return
+				} catch (error) {
+					if (error instanceof RefUpdateError) {
+						console.warn('Pierre CAS conflict, retrying:', error.message)
+						this.pierreState = null
+						continue
+					}
+					throw error
+				}
+			}
+			console.error('Pierre: exhausted CAS retries')
 		} catch (error) {
-			// Log but don't fail the main persist operation
 			console.error('Failed to persist to Pierre:', error)
 			this.reportError(error)
 		}
@@ -1368,5 +1458,15 @@ async function listAllObjectKeys(bucket: R2Bucket, prefix: string): Promise<stri
 const PERSIST_RETRIES_NOTIFY_THRESHOLD = 10
 const PERSIST_RETRIES_MAX = 100
 
-const SNAPSHOT_FILE_NAME = 'snapshot.json'
 const PIERRE_AUTHOR = { email: 'huppy@tldraw.com', name: 'huppy [bot]' }
+
+interface PierreState {
+	headSha: string | undefined
+	documentClock: number
+}
+
+/** Shape of meta.json stored in Pierre archives. */
+export interface PierreMeta {
+	documentClock: number
+	schema: RoomSnapshot['schema']
+}

apps/dotcom/sync-worker/src/routes/getPierreHistory.ts

@@ -32,7 +32,7 @@ export async function getPierreHistory(
 
 	try {
 		const envName = env.TLDRAW_ENV || 'development'
-		const repoId = `${envName}/snapshots/${roomId}`
+		const repoId = `${envName}/files/${roomId}`
 
 		const repo = await pierreClient.findOne({ id: repoId })
 		if (!repo) {

apps/dotcom/sync-worker/src/routes/getPierreHistorySnapshot.ts

@@ -2,6 +2,7 @@ import { notFound } from '@tldraw/worker-shared'
 import { IRequest } from 'itty-router'
 import { Environment } from '../types'
 import { createPierreClient } from '../utils/createPierreClient'
+import { reconstructSnapshotFromPierre } from '../utils/pierreSnapshot'
 import { isRoomIdTooLong, roomIdIsTooLong } from '../utils/roomIdIsTooLong'
 import { requireWriteAccessToFile } from '../utils/tla/getAuth'
 import { isTestFile } from '../utils/tla/isTestFile'
@@ -24,7 +25,7 @@ export async function getPierreHistorySnapshot(
 		return new Response('Not found', { status: 404 })
 	}
 
-	const commitHash = request.params.timestamp // This is actually the commit hash now
+	const commitHash = request.params.timestamp
 	if (!commitHash) {
 		return new Response('Missing commit hash', { status: 400 })
 	}
@@ -36,21 +37,18 @@ export async function getPierreHistorySnapshot(
 
 	try {
 		const envName = env.TLDRAW_ENV || 'development'
-		const repoId = `${envName}/snapshots/${roomId}`
+		const repoId = `${envName}/files/${roomId}`
 
 		const repo = await pierreClient.findOne({ id: repoId })
 		if (!repo) {
 			return new Response('Not found', { status: 404 })
 		}
 
-		// Get the file stream and return it
-		const fileStream = await repo.getFileStream({ path: 'snapshot.json', ref: commitHash })
+		const snapshot = await reconstructSnapshotFromPierre(repo, commitHash)
 
-		// The content should be the snapshot JSON
-		return new Response(fileStream.body, {
+		return new Response(JSON.stringify(snapshot), {
 			headers: {
 				'content-type': 'application/json',
-				// cache forever
 				'Cache-Control': 'public, max-age=31536000, immutable',
 			},
 		})

apps/dotcom/sync-worker/src/utils/pierreSnapshot.ts

@@ -0,0 +1,65 @@
+import type { Repo } from '@pierre/storage'
+import { RoomSnapshot } from '@tldraw/sync-core'
+import { createTarDecoder } from 'modern-tar'
+import type { PierreMeta } from '../TLDrawDurableObject'
+
+/** Actual format from Pierre: listFiles and getArchiveStream use repo-root paths with no prefix. */
+function isMetaJson(name: string) {
+	return name === 'meta.json'
+}
+
+/** Record files are exactly records/<id>.json; tar also includes a "records/" directory entry (excluded here). */
+function isRecordFile(name: string) {
+	return /^records\/[^/]+\.json$/.test(name)
+}
+
+/**
+ * Reconstruct a RoomSnapshot from a Pierre repo at a given ref.
+ * Streams the tar archive — only the parsed JSON values are held in memory.
+ */
+export async function reconstructSnapshotFromPierre(
+	repo: Repo,
+	ref: string
+): Promise<RoomSnapshot> {
+	const resp = await repo.getArchiveStream({ ref })
+	if (!resp.body) {
+		throw new Error(`Empty archive body from Pierre at ref ${ref}`)
+	}
+	const entries = resp.body
+		.pipeThrough(new DecompressionStream('gzip'))
+		.pipeThrough(createTarDecoder())
+
+	let meta: PierreMeta | null = null
+	const documents: RoomSnapshot['documents'] = []
+
+	const reader = entries.getReader()
+	try {
+		while (true) {
+			const { done, value: entry } = await reader.read()
+			if (done) break
+			const name = entry.header.name
+			if (isMetaJson(name)) {
+				meta = JSON.parse(await new Response(entry.body).text()) as PierreMeta
+			} else if (isRecordFile(name)) {
+				const state = JSON.parse(
+					await new Response(entry.body).text()
+				) as RoomSnapshot['documents'][number]['state']
+				documents.push({ state, lastChangedClock: 0 })
+			} else {
+				await entry.body.cancel()
+			}
+		}
+	} finally {
+		reader.releaseLock()
+	}
+
+	if (!meta) {
+		throw new Error(`No meta.json found in Pierre archive at ref ${ref}`)
+	}
+
+	return {
+		documentClock: meta.documentClock,
+		schema: meta.schema,
+		documents,
+	}
+}