From 1844b385fac49e990664bcefec7784b29bf77274 Mon Sep 17 00:00:00 2001 From: Adrian Webb Date: Mon, 31 Aug 2026 21:16:51 -0400 Subject: [PATCH] Complete live library head reconciliation --- .../knowledge/gateway-treedx-connection.ts | 4 +- .../claims/claim-platform-operation.ts | 6 ++- .../remote-head-reconciliation-executor.ts | 23 ++++------- .../remote-head-reconciliation-scheduler.ts | 8 +++- .../treedx/remote-head-reconciliation.test.ts | 38 ++++++++++++------- 5 files changed, 45 insertions(+), 34 deletions(-) diff --git a/src/api/knowledge/gateway-treedx-connection.ts b/src/api/knowledge/gateway-treedx-connection.ts index 5dd4189b..92a74009 100644 --- a/src/api/knowledge/gateway-treedx-connection.ts +++ b/src/api/knowledge/gateway-treedx-connection.ts @@ -95,13 +95,15 @@ export async function resolveKnowledgeGatewayConnection(store: any, input: { scope: { repositoryIds: [repositoryId], capabilities: input.replicationRefs?.length ? ['repos:read', 'files:read', 'git:read', 'git:fetch', 'git:push', 'registry:read', 'snapshot:build', 'artifact:export'] : input.maintenanceRefs?.length - ? ['repos:read', 'files:read', 'git:read', 'git:diff', 'git:fetch', 'git:push', 'registry:read', 'policy:write'] + ? ['repos:read', 'files:read', 'files:search', 'git:read', 'git:diff', 'git:fetch', 'git:push', + 'registry:read', 'graph:query', 'graph:refresh', 'policy:write'] : input.publishRefs?.length ? ['repos:read', 'files:read', 'files:search', 'git:read', 'git:fetch', 'git:push', 'registry:read', 'graph:query', 'graph:refresh'] : input.write ? ['repos:read', 'repos:write', 'workspace:create', 'files:read', 'files:search', 'files:write', 'files:delete', 'git:read', 'git:diff', 'git:commit', 'graph:query', 'graph:refresh'] : ['repos:read', 'files:read', 'files:search', 'git:read', 'git:diff', 'graph:query'], refs: [...new Set([text(library.contentRepositoryRef, library.contentRepositoryDefaultBranch, 'main'), + canonicalTreeDxBranchRef(library.contentRepositoryDefaultBranch ?? 'main'), ...(input.write || input.communicationPaths || input.authoringPaths ? [canonicalAuthoringRef] : []), ...(input.readRefs ?? []), ...(input.publishRefs ?? []), ...(input.maintenanceRefs ?? []), ...(input.replicationRefs ?? []), ...(input.workspaceRefs ?? [])])], diff --git a/src/api/store/operations/lifecycle/claims/claim-platform-operation.ts b/src/api/store/operations/lifecycle/claims/claim-platform-operation.ts index 56974014..c0820768 100644 --- a/src/api/store/operations/lifecycle/claims/claim-platform-operation.ts +++ b/src/api/store/operations/lifecycle/claims/claim-platform-operation.ts @@ -17,7 +17,8 @@ export async function claimPlatformOperationMethod(this: ControlPlaneStore, inpu OR (status IN ('leased', 'running') AND lease_expires_at IS NOT NULL AND lease_expires_at < ?) ) ${capabilityWhere} - ORDER BY CASE namespace WHEN 'knowledge' THEN 0 WHEN 'feedback' THEN 1 ELSE 2 END, + ORDER BY CASE WHEN namespace='knowledge' THEN 0 WHEN namespace='feedback' THEN 1 + WHEN namespace='treedx' AND operation='reconcile_remote_head' THEN 2 ELSE 3 END, created_at ASC LIMIT ?`, [input.operationId, now, ...capabilities, limit]) : await this.all(`SELECT * FROM platform_operations WHERE ( @@ -25,7 +26,8 @@ export async function claimPlatformOperationMethod(this: ControlPlaneStore, inpu OR (status IN ('leased', 'running') AND lease_expires_at IS NOT NULL AND lease_expires_at < ?) ) ${capabilityWhere} - ORDER BY CASE namespace WHEN 'knowledge' THEN 0 WHEN 'feedback' THEN 1 ELSE 2 END, + ORDER BY CASE WHEN namespace='knowledge' THEN 0 WHEN namespace='feedback' THEN 1 + WHEN namespace='treedx' AND operation='reconcile_remote_head' THEN 2 ELSE 3 END, created_at ASC LIMIT ?`, [now, ...capabilities, limit]); const row = rows[0]; if (!row) diff --git a/src/operations-runner/treedx/remote-head-reconciliation-executor.ts b/src/operations-runner/treedx/remote-head-reconciliation-executor.ts index cd7cd0f7..e777f7ae 100644 --- a/src/operations-runner/treedx/remote-head-reconciliation-executor.ts +++ b/src/operations-runner/treedx/remote-head-reconciliation-executor.ts @@ -68,23 +68,14 @@ export function createTreeDxRemoteHeadReconciliationExecutor(options: any) { await connection.client.fetchRemote({ repoId: connection.repositoryId, remoteName: 'origin', remoteUrl: binding.clone_url, credentialId: delivery.deliveryId, refspecs: [`+${publicationRef}:${remoteRef}`] }); - let repositoryRefs = refs(await connection.client.upstream.repositories.refs(connection.repositoryId)); + const repositoryRefs = refs(await connection.client.upstream.repositories.refs(connection.repositoryId)); if (head(repositoryRefs, remoteRef) !== remoteHead) throw new Error('TreeDX did not fetch the protected branch head exactly.'); - const beforeHead = head(repositoryRefs, publicationRef); - let promotion: any = { status: 'already_current', beforeHead, afterHead: beforeHead }; - if (beforeHead !== remoteHead) { - promotion = await connection.client.promoteRef({ repoId: connection.repositoryId, sourceRef: remoteRef, - destinationRef: publicationRef, expectedDestinationHead: beforeHead }); - if (promotion.afterHead !== remoteHead) throw new Error('TreeDX did not advance its current logical view to the protected branch head.'); - } - repositoryRefs = refs(await connection.client.upstream.repositories.refs(connection.repositoryId)); - if (head(repositoryRefs, publicationRef) !== remoteHead) throw new Error('TreeDX current logical view failed exact read-back verification.'); const graph = await completedGraphRefresh(connection.client, { repoId: connection.repositoryId, - ref: publicationRef, paths: ['**'], forceFull: true }); + ref: remoteRef, paths: ['**'], forceFull: true }); await connection.client.refreshSearchIndex({ repoId: connection.repositoryId, - ref: publicationRef, paths: ['**'], incremental: false }); + ref: remoteRef, paths: ['**'], incremental: false }); const search = result(await connection.client.upstream.searchIndex.status(connection.repositoryId, - { ref: publicationRef }), 'index'); + { ref: remoteRef }), 'index'); if (String(graph.resolvedRef ?? remoteHead) !== remoteHead || String(search.resolvedRef ?? '') !== remoteHead || search.ready !== true || search.stale === true || !(Number(search.segmentCount) > 0)) { throw new Error('TreeDX graph/search did not converge on the protected branch head.'); @@ -98,16 +89,16 @@ export function createTreeDxRemoteHeadReconciliationExecutor(options: any) { repositoryId: connection.repositoryId, contentPath: library.contentPath, contentRepositoryUrl: library.contentRepositoryUrl, contentRepositoryDefaultBranch: library.contentRepositoryDefaultBranch, - contentRepositoryRef: publicationRef, metadata, + contentRepositoryRef: remoteRef, metadata, }); if (!updated) throw new Error('TreeDX library binding could not be updated after reconciliation.'); await store.run(`UPDATE project_remote_repository_bindings SET expected_head=?,observed_head=?,drift='none', version=version+1,updated_at=? WHERE id=?`, [remoteHead, remoteHead, now, binding.id]); const replication = await enqueueTreeDxCommitReplication(store, { teamId, projectId, commitSha: remoteHead, - sourceRef: publicationRef, createdAt: now }); + sourceRef: remoteRef, createdAt: now }); await context.checkpoint({ phase: 'treedx.remote-head.reconciled', projectId, remoteHead }, { kind: 'treedx.remote-head.reconciled', data: { projectId, remoteHead, publicationRef } }); - return { projectId, publicationRef, remoteHead, promotion, graph, search, replication }; + return { projectId, publicationRef, currentLogicalRef: remoteRef, remoteHead, graph, search, replication }; }, }; } diff --git a/src/operations-runner/treedx/remote-head-reconciliation-scheduler.ts b/src/operations-runner/treedx/remote-head-reconciliation-scheduler.ts index acf491d9..db5d2bb8 100644 --- a/src/operations-runner/treedx/remote-head-reconciliation-scheduler.ts +++ b/src/operations-runner/treedx/remote-head-reconciliation-scheduler.ts @@ -44,13 +44,17 @@ export class TreeDxRemoteHeadReconciliationScheduler { const metadata = parse(binding.metadata_json); const currentResolvedRef = String(metadata.resolvedRef ?? ''); const canonicalRef = String(binding.content_repository_ref ?? ''); + const remoteRef = `refs/remotes/origin/${publicationRef.slice('refs/heads/'.length)}`; if (binding.expected_head === remoteHead && binding.observed_head === remoteHead - && currentResolvedRef === remoteHead && canonicalRef === publicationRef) continue; + && currentResolvedRef === remoteHead && canonicalRef === remoteRef) continue; const idempotencyKey = operationKey(String(binding.project_id), publicationRef, remoteHead); - await this.store.createPlatformOperation({ namespace: 'treedx', operation: 'reconcile_remote_head', + const operation = await this.store.createPlatformOperation({ namespace: 'treedx', operation: 'reconcile_remote_head', target: 'control_plane_operations_runner', idempotencyKey, input: { teamId: binding.team_id, projectId: binding.project_id, publicationRef, remoteHead }, requestedByType: 'service', requestedById: 'treedx-remote-head-reconciliation-scheduler' }); + if (['failed', 'cancelled'].includes(String(operation?.status ?? ''))) { + await this.store.retryPlatformOperation(operation.id); + } queued += 1; } catch { // A broken provider binding must not block reconciliation for other projects. diff --git a/tests/unit/control-plane/treedx/remote-head-reconciliation.test.ts b/tests/unit/control-plane/treedx/remote-head-reconciliation.test.ts index eccc837c..f1d7e0c1 100644 --- a/tests/unit/control-plane/treedx/remote-head-reconciliation.test.ts +++ b/tests/unit/control-plane/treedx/remote-head-reconciliation.test.ts @@ -46,7 +46,7 @@ describe('TreeDX protected branch reconciliation', () => { owner: 'treeseed-ai', name: 'sdk-library', publication_ref: publicationRef, expected_head: newHead, observed_head: newHead, content_repository_ref: oldHead, metadata_json: JSON.stringify({ resolvedRef: oldHead }) }]; }, - async createPlatformOperation(value: any) { operations.push(value); }, + async createPlatformOperation(value: any) { operations.push(value); return { id: 'operation', status: 'queued' }; }, }; const scheduler = new TreeDxRemoteHeadReconciliationScheduler(store, 1, githubFetch()); const first = await scheduler.runIfDue(Date.parse('2026-08-31T12:00:00.000Z')); @@ -60,26 +60,39 @@ describe('TreeDX protected branch reconciliation', () => { const operations: any[] = []; const store: any = { config: {}, async all() { return [{ id: 'binding', team_id: 'team', project_id: 'project', authority_id: 'authority', owner: 'treeseed-ai', name: 'sdk-library', publication_ref: publicationRef, - expected_head: newHead, observed_head: newHead, content_repository_ref: publicationRef, + expected_head: newHead, observed_head: newHead, content_repository_ref: 'refs/remotes/origin/staging', metadata_json: JSON.stringify({ resolvedRef: newHead }) }]; }, - async createPlatformOperation(value: any) { operations.push(value); } }; + async createPlatformOperation(value: any) { operations.push(value); return { id: 'operation', status: 'queued' }; } }; const result = await new TreeDxRemoteHeadReconciliationScheduler(store, 1, githubFetch()) .runIfDue(Date.parse('2026-08-31T12:00:00.000Z')); expect(result).toMatchObject({ queued: 0, failed: 0 }); expect(operations).toHaveLength(0); }); - it('fetches, promotes, indexes, advances the logical binding, and queues the exact R2 mirror', async () => { + it('retries the same exact-head operation after a recoverable failure', async () => { + const retried: string[] = []; + const store: any = { config: {}, async all() { return [{ id: 'binding', team_id: 'team', project_id: 'project', + authority_id: 'authority', owner: 'treeseed-ai', name: 'sdk-library', publication_ref: publicationRef, + expected_head: oldHead, observed_head: oldHead, content_repository_ref: publicationRef, + metadata_json: JSON.stringify({ resolvedRef: oldHead }) }]; }, + async createPlatformOperation() { return { id: 'failed-operation', status: 'failed' }; }, + async retryPlatformOperation(id: string) { retried.push(id); } }; + const result = await new TreeDxRemoteHeadReconciliationScheduler(store, 1, githubFetch()) + .runIfDue(Date.parse('2026-08-31T12:00:00.000Z')); + expect(result).toMatchObject({ queued: 1, failed: 0 }); + expect(retried).toEqual(['failed-operation']); + }); + + it('fetches, indexes, advances the logical binding, and queues the exact R2 mirror', async () => { const runs: Array<{ query: string; params: unknown[] }> = []; const upserts: any[] = []; - let refs = [{ name: publicationRef, target: oldHead }, { name: 'refs/remotes/origin/staging', target: newHead }]; + const remoteRef = 'refs/remotes/origin/staging'; + const refs = [{ name: publicationRef, target: oldHead }, { name: remoteRef, target: newHead }]; const client: any = { fetchRemote: vi.fn(async () => ({})), upstream: { repositories: { refs: vi.fn(async () => ({ refs })) }, searchIndex: { status: vi.fn(async () => ({ index: { ready: true, stale: false, resolvedRef: newHead, segmentCount: 4 } })) } }, - promoteRef: vi.fn(async () => { refs = [{ name: publicationRef, target: newHead }, - { name: 'refs/remotes/origin/staging', target: newHead }]; return { beforeHead: oldHead, afterHead: newHead }; }), refreshGraph: vi.fn(async () => ({ graph: { status: 'completed', resolvedRef: newHead, graphVersion: 'graph-1' } })), refreshSearchIndex: vi.fn(async () => ({ index: { status: 'completed' } })), getPlacement: vi.fn(async () => ({ primaryNodeId: 'node' })), @@ -99,15 +112,14 @@ describe('TreeDX protected branch reconciliation', () => { const executor = createTreeDxRemoteHeadReconciliationExecutor({ controlPlaneStore: store, fetchImpl: githubFetch() }); const result = await executor.run({ teamId: 'team', projectId: 'project', publicationRef, remoteHead: newHead }, { operation: { id: 'operation' }, checkpoint: async (...values: any[]) => checkpoints.push(values) }); - expect(client.fetchRemote).toHaveBeenCalledWith(expect.objectContaining({ refspecs: [`+${publicationRef}:refs/remotes/origin/staging`] })); - expect(client.promoteRef).toHaveBeenCalledWith(expect.objectContaining({ expectedDestinationHead: oldHead })); - expect(client.refreshGraph).toHaveBeenCalledWith(expect.objectContaining({ ref: publicationRef, forceFull: true })); - expect(client.refreshSearchIndex).toHaveBeenCalledWith(expect.objectContaining({ ref: publicationRef, incremental: false })); - expect(upserts[0]).toMatchObject({ contentRepositoryRef: publicationRef, + expect(client.fetchRemote).toHaveBeenCalledWith(expect.objectContaining({ refspecs: [`+${publicationRef}:${remoteRef}`] })); + expect(client.refreshGraph).toHaveBeenCalledWith(expect.objectContaining({ ref: remoteRef, forceFull: true })); + expect(client.refreshSearchIndex).toHaveBeenCalledWith(expect.objectContaining({ ref: remoteRef, incremental: false })); + expect(upserts[0]).toMatchObject({ contentRepositoryRef: remoteRef, metadata: { retained: true, resolvedRef: newHead } }); expect(runs.some((entry) => entry.query.includes('UPDATE project_remote_repository_bindings'))).toBe(true); expect(enqueueReplication).toHaveBeenCalledWith(store, expect.objectContaining({ - teamId: 'team', projectId: 'project', commitSha: newHead, sourceRef: publicationRef, + teamId: 'team', projectId: 'project', commitSha: newHead, sourceRef: remoteRef, })); expect(checkpoints).toHaveLength(1); expect(result).toMatchObject({ projectId: 'project', remoteHead: newHead,