From 7401705d88602d00ab5189a8f3d11137196655e2 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 2 Sep 2026 02:16:05 +0000 Subject: [PATCH 1/2] feat(service-datasource,service-cluster): fan datasource record writes out to peer replicas (#13805) The ObjectQL driver registry had no cluster propagation in either direction: each replica filled it at boot from the shared datasource records and mutated it only for the writes it served, so after DELETE /api/v1/datasources/:name only the serving replica evicted the stuck driver (#13578) and the other N-1 kept draining /api/v1/ready until restart. Ruled shape (2026-09-01): the driver registry adopts the cluster-invalidation family metadata.mutated established -- no second mechanism, no poll loop, no delete-only broadcast. DatasourceAdminService publishes the record's ADDRESS on datasource.mutated after create, update and delete (symmetric); peers converge their live pool from their OWN read of the durable sys_metadata row through the pool seams they already own (build / rebuild-in-place / evict / leave alone). Only IPubSub from spec/contracts crosses the new attach seam; service-datasource takes no cluster dependency and objectql is handed no bus. MetadataClusterBridgePlugin late-binds the seam as a third independent lane, guarded off the in-process memory driver like the other two. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01AUF1NoViznQK32gqpK8wS8 --- .../datasource-mutation-cluster-fanout.md | 63 ++ .../metadata-cluster-bridge-plugin.test.ts | 134 +++- .../src/metadata-cluster-bridge-plugin.ts | 85 ++- .../datasource-cluster-convergence.test.ts | 691 ++++++++++++++++++ .../src/datasource-admin-plugin.ts | 113 +++ .../src/datasource-admin-service.ts | 204 ++++++ .../services/service-datasource/src/index.ts | 8 + scripts/engine-double-contract.pinned.json | 15 + 8 files changed, 1309 insertions(+), 4 deletions(-) create mode 100644 .changeset/datasource-mutation-cluster-fanout.md create mode 100644 packages/services/service-datasource/src/__tests__/datasource-cluster-convergence.test.ts diff --git a/.changeset/datasource-mutation-cluster-fanout.md b/.changeset/datasource-mutation-cluster-fanout.md new file mode 100644 index 0000000000..dda5095e13 --- /dev/null +++ b/.changeset/datasource-mutation-cluster-fanout.md @@ -0,0 +1,63 @@ +--- +"@objectstack/service-datasource": minor +"@objectstack/service-cluster": minor +--- + +feat(service-datasource,service-cluster): fan datasource record writes out to peer replicas — a deleted datasource no longer keeps draining `/api/v1/ready` on every replica that did not serve the DELETE (#13805) + +Measured on a live 3-replica EE deployment: the ObjectQL DRIVER registry had +no cluster propagation in either direction. Each replica filled it at boot +from the shared datasource records and mutated it only for the writes IT +served, so after `DELETE /api/v1/datasources/:name` only the replica that +served the DELETE evicted the stuck driver (#13578's door) — the other N-1 +kept it, and `/api/v1/ready` kept answering 503 there, until restart. A +datasource created through one replica likewise had no pool on any other +until restart. + +Maintainer-ruled design (2026-09-01): the driver registry adopts the same +cluster-invalidation family `metadata.mutated` (#13331) established — no +second propagation mechanism, no bespoke poll loop, and no delete-only +broadcast (that would have made delete more cluster-aware than create, a new +asymmetry rather than a repair). + +- **Symmetric publisher at the three write doors.** `DatasourceAdminService` + now publishes the record's ADDRESS on a new cluster channel + `datasource.mutated` (`DATASOURCE_MUTATION_CLUSTER_CHANNEL`, payload + `ClusterDatasourceMutationPayload` — `{ originNode?, name }`) after + `createDatasource`, `updateDatasource` and `removeDatasource`. Fire-and- + forget: a publish failure never fails the write it announces. + `migrateCredential` does not publish — it leaves the live pool alone by + design, on every replica alike. +- **Peers converge from their own read of the SHARED record.** On receipt a + replica re-reads the durable `sys_metadata` row for that name — the same + store its boot rehydration reads, not its per-replica metadata registry — + and converges its live pool through the seams it already owns: builds what + is missing, rebuilds in place what changed (`reregisterPool`, keeping the + old pool on failure exactly as the serving replica's update path does), + evicts what is gone (`unregisterPool` → the #13578 eviction door), and + leaves a matching pool untouched. The payload is a signal, never trusted + content, so a duplicate or re-ordered delivery converges to the same pool + state by construction — which is what makes a replayed create safe without + any new idempotency machinery. A name the replica never pooled is left + alone, so a stray signal cannot reach a code-defined pool. +- **New attach seam, mirrored from the shipped bridges.** + `DatasourceAdminService.attachDatasourceMutationPubSub(pubsub, nodeId)` — + idempotent on the `(pubsub, nodeId)` pair, loopback suppression via + `originNode`, shaped after the protocol's `attachMetadataMutationPubSub()`. + Only `IPubSub` from `@objectstack/spec/contracts` crosses it: + `@objectstack/service-datasource` takes no dependency on the cluster + service, and `@objectstack/objectql` — the registry's owner — is handed no + bus. The host wires the receive half through a new optional + `DatasourceAdminServiceConfig.convergePool` seam; `DatasourceAdminServicePlugin` + supplies it. +- **`MetadataClusterBridgePlugin` gains a third, independent lane** that + late-binds the seam at `kernel:ready` beside the metadata-service and + protocol lanes, duck-typed on the `datasource-admin` service. It skips the + in-process memory driver (nothing to fan out to), the guard the other lanes + carry, so a single-replica boot behaves byte-identically to before. + +No shipped driver exceeds at-most-once delivery, so a lost message still +degrades to the pre-existing bound (the next boot's full rehydration); this +channel narrows the window from "until every replica restarts" to one network +hop. The `/api/v1/meta/datasource` metadata registry's own cross-replica +coherence (#13609) is a different sink and is not touched here. diff --git a/packages/services/service-cluster/src/metadata-cluster-bridge-plugin.test.ts b/packages/services/service-cluster/src/metadata-cluster-bridge-plugin.test.ts index e1258145a8..fc4cfbcb23 100644 --- a/packages/services/service-cluster/src/metadata-cluster-bridge-plugin.test.ts +++ b/packages/services/service-cluster/src/metadata-cluster-bridge-plugin.test.ts @@ -48,10 +48,22 @@ interface HarnessOptions { * `'real'` (exposes attachMetadataMutationPubSub). */ protocol?: 'none' | 'bare' | 'real'; + /** + * [#13805] The `datasource-admin` slot: `'none'` (getService throws — the + * default, so the #13331 cases above read exactly as they did), `'bare'` + * (present, no attachDatasourceMutationPubSub — an older implementation), + * or `'real'` (exposes attachDatasourceMutationPubSub). + */ + datasourceAdmin?: 'none' | 'bare' | 'real'; + /** When true, the datasource-admin seam throws on attach. */ + datasourceAttachThrows?: boolean; } function makeHarness(opts: HarnessOptions = {}) { - const { driver = 'redis', metadata = 'fallback', protocol = 'real' } = opts; + const { + driver = 'redis', metadata = 'fallback', protocol = 'real', + datasourceAdmin = 'none', datasourceAttachThrows = false, + } = opts; const logger = { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }; @@ -59,6 +71,11 @@ function makeHarness(opts: HarnessOptions = {}) { const attachMetadata = vi.fn((_pubsub: unknown, _nodeId: string) => detachMetadata); const detachMutation = vi.fn(); const attachMutation = vi.fn((_pubsub: unknown, _nodeId: string) => detachMutation); + const detachDatasource = vi.fn(); + const attachDatasource = vi.fn((_pubsub: unknown, _nodeId: string) => { + if (datasourceAttachThrows) throw new Error('datasource attach exploded'); + return detachDatasource; + }); const pubsub = { publish: vi.fn(), subscribe: vi.fn(), close: vi.fn() }; const cluster = @@ -74,6 +91,10 @@ function makeHarness(opts: HarnessOptions = {}) { protocol === 'none' ? undefined : protocol === 'bare' ? { saveMetaItem: vi.fn() } : { attachMetadataMutationPubSub: attachMutation }; + const datasourceAdminService = + datasourceAdmin === 'none' ? undefined + : datasourceAdmin === 'bare' ? { listDatasources: vi.fn() } + : { attachDatasourceMutationPubSub: attachDatasource }; const hooks = new Map Promise | void>>(); const ctx = { @@ -96,6 +117,10 @@ function makeHarness(opts: HarnessOptions = {}) { if (!protocolService) throw new Error('service not found: protocol'); return protocolService; } + if (name === 'datasource-admin') { + if (!datasourceAdminService) throw new Error('service not found: datasource-admin'); + return datasourceAdminService; + } throw new Error(`service not found: ${name}`); }, } as unknown as PluginContext; @@ -107,6 +132,7 @@ function makeHarness(opts: HarnessOptions = {}) { return { ctx, logger, fire, pubsub, attachMetadata, detachMetadata, attachMutation, detachMutation, + attachDatasource, detachDatasource, }; } @@ -280,3 +306,109 @@ describe('[#14021] lane 1 — an in-process bus must not be reported as “bridg expect(h.logger.error).not.toHaveBeenCalled(); }); }); + +describe('[#13805] lane 3 — the datasource admin service’s datasource.mutated fan-out', () => { + it('⭐ attaches on a cross-process driver and reports it — independently of lanes 1 and 2', async () => { + // The shipped EE shape again, one owner over: no manager-backed + // metadata slot, no protocol seam, and a real datasource-admin service. + // Lane 3 must attach exactly there, with nothing from the other two + // lanes taking it down. + const h = makeHarness({ driver: 'redis', metadata: 'none', protocol: 'none', datasourceAdmin: 'real' }); + await new MetadataClusterBridgePlugin().init(h.ctx); + await h.fire('kernel:ready'); + + expect(h.attachDatasource).toHaveBeenCalledTimes(1); + expect(h.attachDatasource).toHaveBeenCalledWith(h.pubsub, 'node-a'); + // Asserted VERBATIM, like lane 1's and lane 2's lines: the wording is + // what an operator reads as "datasource fan-out is on". + expect(infoLines(h)).toContain( + 'MetadataClusterBridgePlugin: bridged datasource.mutated → cluster.pubsub (node=node-a)', + ); + expect(h.logger.error).not.toHaveBeenCalled(); + }); + + it('all three lanes attach together when every owner exposes its seam', async () => { + const h = makeHarness({ driver: 'redis', metadata: 'manager', protocol: 'real', datasourceAdmin: 'real' }); + await new MetadataClusterBridgePlugin().init(h.ctx); + await h.fire('kernel:ready'); + + expect(h.attachMetadata).toHaveBeenCalledWith(h.pubsub, 'node-a'); + expect(h.attachMutation).toHaveBeenCalledWith(h.pubsub, 'node-a'); + expect(h.attachDatasource).toHaveBeenCalledWith(h.pubsub, 'node-a'); + expect(warnLines(h)).toEqual([]); + }); + + it('skips attach on the in-process memory driver — no peers to reach, nothing said above debug', async () => { + const h = makeHarness({ driver: 'memory', datasourceAdmin: 'real' }); + await new MetadataClusterBridgePlugin().init(h.ctx); + await h.fire('kernel:ready'); + + // The guard lanes 1 and 2 carry, from birth: on the memory driver a + // single replica's behaviour stays byte-identical to the pre-bridge + // one — no subscription, no publisher, no "bridged" claim. + expect(h.attachDatasource).not.toHaveBeenCalled(); + expect(infoLines(h).some((l) => l.includes('datasource.mutated'))).toBe(false); + expect( + debugLines(h).some((l) => l.includes('is in-process') && l.includes('datasource fan-out')), + ).toBe(true); + }); + + it('skips quietly when no datasource-admin service is registered', async () => { + const h = makeHarness({ driver: 'redis', datasourceAdmin: 'none' }); + await new MetadataClusterBridgePlugin().init(h.ctx); + await h.fire('kernel:ready'); + + expect(h.attachDatasource).not.toHaveBeenCalled(); + expect(h.logger.error).not.toHaveBeenCalled(); + expect(warnLines(h).some((l) => l.includes('datasource'))).toBe(false); + }); + + it('skips quietly when the service does not expose the seam', async () => { + const h = makeHarness({ driver: 'redis', datasourceAdmin: 'bare' }); + await new MetadataClusterBridgePlugin().init(h.ctx); + await h.fire('kernel:ready'); + + expect(h.attachDatasource).not.toHaveBeenCalled(); + expect(h.logger.error).not.toHaveBeenCalled(); + }); + + it('no cluster service at all skips lane 3 too', async () => { + const h = makeHarness({ driver: null, datasourceAdmin: 'real' }); + await new MetadataClusterBridgePlugin().init(h.ctx); + await h.fire('kernel:ready'); + + expect(h.attachDatasource).not.toHaveBeenCalled(); + }); + + it('a throwing attach is reported and does not take the other lanes down', async () => { + const h = makeHarness({ + driver: 'redis', metadata: 'manager', protocol: 'real', + datasourceAdmin: 'real', datasourceAttachThrows: true, + }); + await new MetadataClusterBridgePlugin().init(h.ctx); + await h.fire('kernel:ready'); + + expect(h.attachMetadata).toHaveBeenCalledTimes(1); + expect(h.attachMutation).toHaveBeenCalledTimes(1); + expect(h.logger.error).toHaveBeenCalledWith( + 'MetadataClusterBridgePlugin: datasource-lane attach failed', + expect.any(Error), + ); + expect(infoLines(h).some((l) => l.includes('datasource.mutated'))).toBe(false); + }); + + it('kernel:shutdown detaches lane 3, and a throwing lane-2 detach does not strand it', async () => { + const h = makeHarness({ driver: 'redis', metadata: 'manager', protocol: 'real', datasourceAdmin: 'real' }); + h.detachMutation.mockImplementation(() => { throw new Error('detach exploded'); }); + await new MetadataClusterBridgePlugin().init(h.ctx); + await h.fire('kernel:ready'); + await h.fire('kernel:shutdown'); + + expect(h.detachDatasource).toHaveBeenCalledTimes(1); + expect(h.logger.error).toHaveBeenCalled(); + + // Idempotent: a second shutdown does not detach twice. + await h.fire('kernel:shutdown'); + expect(h.detachDatasource).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/services/service-cluster/src/metadata-cluster-bridge-plugin.ts b/packages/services/service-cluster/src/metadata-cluster-bridge-plugin.ts index b7d4d18b41..ed29591985 100644 --- a/packages/services/service-cluster/src/metadata-cluster-bridge-plugin.ts +++ b/packages/services/service-cluster/src/metadata-cluster-bridge-plugin.ts @@ -15,8 +15,8 @@ import { isInProcessClusterDriver } from './split-brain-guard.js'; * state-owner packages only need the `IPubSub` interface, which lives in * `@objectstack/spec/contracts`. * - * TWO lanes, late-bound independently at `kernel:ready`, because the state - * that goes stale lives in two different owners (#13331): + * THREE lanes, late-bound independently at `kernel:ready`, because the state + * that goes stale lives in three different owners (#13331, #13805): * * 1. **Metadata service** (`attachClusterPubSub()` — `metadata.changed`): * replays watch events into peer `MetadataManager` caches @@ -34,6 +34,15 @@ import { isInProcessClusterDriver } from './split-brain-guard.js'; * 67×201 / 133×404). The lanes are independent on purpose: the boot * shape that lacks lane 1 (host-config, fallback metadata slot) is * exactly the shipped EE shape that needs lane 2. + * 3. **Datasource admin service** (`attachDatasourceMutationPubSub()` — + * `datasource.mutated`): fans a datasource create / update / delete out + * to peers, which converge their ObjectQL DRIVER registry from their OWN + * read of the shared datasource record. Lane 2's family, adopted by the + * driver registry (#13805, ruled 2026-09-01 — the same bridge shape, a + * symmetric signal, no second propagation mechanism): without it a + * `DELETE /api/v1/datasources/:name` recovered `/api/v1/ready` on the + * one replica that served it, and every other replica kept the stuck + * driver until restart. * * Activates each lane only when the cluster service and that lane's state * owner are present and expose the seam. Late binding is achieved via the @@ -42,7 +51,9 @@ import { isInProcessClusterDriver } from './split-brain-guard.js'; * Channels: `metadata.changed` — payload shape defined by * `ClusterMetadataChangedPayload` in `@objectstack/metadata`; * `metadata.mutated` — payload shape defined by - * `ClusterMetadataMutationPayload` in `@objectstack/metadata-protocol`. + * `ClusterMetadataMutationPayload` in `@objectstack/metadata-protocol`; + * `datasource.mutated` — payload shape defined by + * `ClusterDatasourceMutationPayload` in `@objectstack/service-datasource`. * * See `content/docs/kernel/cluster.mdx` §5. */ @@ -53,6 +64,7 @@ export class MetadataClusterBridgePlugin implements Plugin { private detach?: () => void; private detachMutation?: () => void; + private detachDatasource?: () => void; async init(ctx: PluginContext): Promise { ctx.hook('kernel:ready', async () => { @@ -67,6 +79,7 @@ export class MetadataClusterBridgePlugin implements Plugin { } this.attachMetadataServiceLane(ctx, cluster); this.attachProtocolLane(ctx, cluster); + this.attachDatasourceLane(ctx, cluster); }); ctx.hook('kernel:shutdown', async () => { @@ -88,6 +101,15 @@ export class MetadataClusterBridgePlugin implements Plugin { ); } this.detachMutation = undefined; + try { + this.detachDatasource?.(); + } catch (err) { + ctx.logger.error( + 'MetadataClusterBridgePlugin: datasource-lane detach error', + err as Error, + ); + } + this.detachDatasource = undefined; }); } @@ -218,4 +240,61 @@ export class MetadataClusterBridgePlugin implements Plugin { ); } } + + /** + * Lane 3 — the datasource ADMIN SERVICE's `datasource.mutated` fan-out + * (#13805): the driver registry adopting the family lane 2 established. + * + * Duck-typed exactly like lanes 1 and 2 feature-detect their seams: this + * package must not depend on `@objectstack/service-datasource`, and + * `@objectstack/objectql` — the driver registry's owner — is handed no + * bus at all; the admin service publishes on the write doors it already + * owns and converges its pools through the seams it already injects. + * + * Guarded on {@link isInProcessClusterDriver} from birth, like lane 2: the + * in-process memory driver fans out to nobody, and on that driver a single + * replica's behaviour stays byte-identical to the pre-bridge one. + */ + private attachDatasourceLane(ctx: PluginContext, cluster: IClusterService): void { + let admin: unknown; + try { + admin = ctx.getService('datasource-admin'); + } catch { + ctx.logger.debug( + 'MetadataClusterBridgePlugin: no "datasource-admin" service registered, skipping datasource fan-out', + ); + return; + } + + const attach = (admin as { attachDatasourceMutationPubSub?: unknown }) + .attachDatasourceMutationPubSub; + if (typeof attach !== 'function') { + ctx.logger.debug( + 'MetadataClusterBridgePlugin: datasource-admin service does not expose attachDatasourceMutationPubSub(), skipping datasource fan-out', + ); + return; + } + + if (isInProcessClusterDriver(cluster.driver)) { + ctx.logger.debug( + `MetadataClusterBridgePlugin: cluster driver "${cluster.driver}" is in-process; datasource fan-out has no peers to reach, skipping`, + ); + return; + } + + try { + this.detachDatasource = (attach as ( + pubsub: IClusterService['pubsub'], + nodeId: string, + ) => () => void).call(admin, cluster.pubsub, cluster.nodeId); + ctx.logger.info( + `MetadataClusterBridgePlugin: bridged datasource.mutated → cluster.pubsub (node=${cluster.nodeId})`, + ); + } catch (err) { + ctx.logger.error( + 'MetadataClusterBridgePlugin: datasource-lane attach failed', + err as Error, + ); + } + } } diff --git a/packages/services/service-datasource/src/__tests__/datasource-cluster-convergence.test.ts b/packages/services/service-datasource/src/__tests__/datasource-cluster-convergence.test.ts new file mode 100644 index 0000000000..71a85dc6a5 --- /dev/null +++ b/packages/services/service-datasource/src/__tests__/datasource-cluster-convergence.test.ts @@ -0,0 +1,691 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * [#13805] Cross-replica DRIVER-registry convergence — the datasource admin + * service's `datasource.mutated` fan-out, measured over the topology the card + * describes: two `DatasourceAdminServicePlugin` boots ("replicas") over ONE + * shared `sys_metadata` store, each with its OWN driver registry and its OWN + * in-memory metadata registry, joined by a bus that delivers every publish to + * every subscriber (publisher included — what a real remote driver does, and + * what the `originNode` loopback guard exists for). + * + * The defect this pins, from a live 3-replica EE deployment: after + * `DELETE /api/v1/datasources/:name` only the replica that served the DELETE + * evicted the driver (#13578's door); every other replica kept the stuck + * driver — and `/api/v1/ready` kept naming it — until restart. Symmetrically, + * a datasource created on one replica had no pool on any other until restart. + * + * --------------------------------------------------------------------------- + * Two-arm design, directions declared BEFORE running + * --------------------------------------------------------------------------- + * • Arm B (bridge attached): a write on the writer reaches the peer and the + * peer's pool converges FROM ITS OWN READ of the shared row -> GREEN + * • Arm A (control, no attach): the same write leaves the peer's registry + * exactly as it was — the pre-fix production shape -> GREEN + * (constrains the instrument: Arm B's convergence is the bridge's doing, + * not a harness artifact that shares a registry between the replicas) + * + * Reverse verification, direction declared for the committed tree: removing + * the `publishDatasourceMutation` call from `removeDatasource` turns EXACTLY + * the delete-convergence cases red ("DELETE on the writer evicts … on the + * PEER", the symmetric-payload case's delete leg) while Arm A, the loopback + * case and every single-replica assertion stay green — with no publisher + * nothing crosses the bus, which is indistinguishable from the shipped defect. + */ + +import { describe, it, expect, vi } from 'vitest'; +import { + assertEngineDeleteDispatch, + assertEngineFindOnePredicate, + assertEngineUpdateDispatch, +} from '@objectstack/metadata-core'; +import type { IDataDriver, IPubSub } from '@objectstack/spec/contracts'; +import { DatasourceAdminServicePlugin } from '../datasource-admin-plugin.js'; +import { + DatasourceAdminService, + DATASOURCE_MUTATION_CLUSTER_CHANNEL, + type ClusterDatasourceMutationPayload, + type DatasourceAdminServiceConfig, + type StoredDatasource, +} from '../datasource-admin-service.js'; +import type { + IDatasourceDriverFactory, + DatasourceConnectionSpec, +} from '../contracts/datasource-driver-factory.js'; +import type { DatasourceDraft } from '../contracts/index.js'; + +type Row = Record; + +/** The SHARED database — the one `sys_metadata` every replica writes and reads. */ +function makeSharedStore() { + return { rows: [] as Row[], seq: 0 }; +} + +/** A persisted runtime datasource, in the row shape `persistDatasourceRow` writes. */ +function sysRow(record: StoredDatasource): Row { + const now = new Date().toISOString(); + return { + id: `meta_${record.name}`, + name: record.name, + type: 'datasource', + scope: 'platform', + metadata: JSON.stringify(record), + state: 'active', + version: 1, + created_at: now, + updated_at: now, + }; +} + +function matches(row: Row, where: Record): boolean { + for (const [k, v] of Object.entries(where)) { + if (k.startsWith('$')) throw new Error(`fake engine: unsupported operator ${k}`); + if (v === undefined) continue; + if (row[k] !== v) return false; + } + return true; +} + +/** The fake driver a pool is: it knows what it was built from and whether its connection is open. */ +interface FakeDriver { + name: string; + builtFrom: DatasourceConnectionSpec & { secret?: string }; + connected: boolean; + closed: boolean; + disconnect: () => Promise; +} + +/** A fake factory whose drivers capture the spec they were built from. */ +function fakeFactory( + opts: { failWhen?: (spec: DatasourceConnectionSpec & { secret?: string }) => boolean } = {}, +) { + const created: Array = []; + const factory: IDatasourceDriverFactory & { created: typeof created } = { + created, + supports: () => true, + create: async (spec) => { + const builtFrom = spec as DatasourceConnectionSpec & { secret?: string }; + created.push(builtFrom); + const driver: FakeDriver = { + name: 'com.fake.driver', + builtFrom, + connected: false, + closed: false, + disconnect: async () => { + driver.closed = true; + }, + }; + return { + driver, + connect: async () => { + if (opts.failWhen?.(builtFrom)) throw new Error('connection refused'); + driver.connected = true; + }, + }; + }, + }; + return factory; +} + +/** + * A remote-driver-shaped bus: one transport object, every publish delivered + * synchronously to EVERY subscription — the publisher's own node included. + */ +function makeBus() { + type Msg = { channel: string; payload: unknown; publishedAt: number }; + const subs: Array<{ channel: string; handler: (msg: Msg) => void }> = []; + const published: Array<{ channel: string; payload: ClusterDatasourceMutationPayload }> = []; + const bus = { + async publish(channel: string, payload: unknown) { + published.push({ channel, payload: payload as ClusterDatasourceMutationPayload }); + for (const s of [...subs]) { + if (s.channel === channel) s.handler({ channel, payload, publishedAt: Date.now() }); + } + }, + subscribe(channel: string, handler: (msg: never) => void) { + const sub = { channel, handler: handler as (msg: Msg) => void }; + subs.push(sub); + return () => { + const i = subs.indexOf(sub); + if (i >= 0) subs.splice(i, 1); + }; + }, + async close() {}, + }; + return { bus: bus as unknown as IPubSub, raw: bus, published, subscriptionCount: () => subs.length }; +} + +/** + * One "replica": the REAL plugin booted over a fake `data` engine whose + * sys_metadata slice is the SHARED store and whose driver-registry slice is + * its own — mirroring the real registry's semantics (`registerDriver` keeps + * the incumbent on a name collision; `unregisterDriver` removes driver + def + * together), because those two behaviours are why convergence needs the + * rebuild primitive rather than a bare re-register. + */ +async function makeReplica( + store: ReturnType, + opts: { failWhen?: (spec: DatasourceConnectionSpec & { secret?: string }) => boolean } = {}, +) { + const drivers = new Map(); + const defs = new Map(); + const evicted: string[] = []; + + const engine = { + // ── the sys_metadata slice (`DataEngineLike`): over the SHARED rows ── + findOne: async (table: string, q: { where: Record }) => { + assertEngineFindOnePredicate(table, q); + return store.rows.find((r) => matches(r, q.where)) ?? null; + }, + find: async (_table: string, q?: { where?: Record; limit?: number }) => { + // Hold the caller's bound AFTER the filter, by PRESENCE — the + // objectql-double-limit contract. + const all = store.rows.filter((r) => matches(r, q?.where ?? {})); + return typeof q?.limit === 'number' ? all.slice(0, q.limit) : all; + }, + insert: async (_table: string, data: Row) => { + const row = { ...data, id: data.id ?? `r_${++store.seq}` }; + store.rows.push(row); + return { id: row.id }; + }, + update: async (_table: string, data: Row, q: { where: Record }) => { + assertEngineUpdateDispatch(data, q); + for (const r of store.rows) if (matches(r, q.where)) Object.assign(r, data); + return {}; + }, + delete: async (_table: string, q: { where: Record }) => { + assertEngineDeleteDispatch(q); + const before = store.rows.length; + for (let i = store.rows.length - 1; i >= 0; i--) { + if (matches(store.rows[i], q.where)) store.rows.splice(i, 1); + } + return { deleted: before - store.rows.length }; + }, + // ── the driver-registry slice (`ConnectionEngineLike`): PER REPLICA ── + registerDriver: (driver: IDataDriver) => { + if (drivers.has(driver.name)) return; + drivers.set(driver.name, driver as unknown as FakeDriver); + }, + unregisterDriver: (name: string) => { + defs.delete(name); + evicted.push(name); + return drivers.delete(name); + }, + getDriverByName: (name: string) => drivers.get(name) as unknown as IDataDriver | undefined, + registerDatasourceDef: (def: { name: string; schemaMode?: string; external?: unknown }) => { + defs.set(def.name, def); + }, + markDatasourceUnavailable: () => {}, + clearDatasourceUnavailable: () => {}, + }; + + // This replica's OWN in-memory metadata registry — per-replica state on + // the host-config boot, which is exactly why convergence must not read it. + const registry = new Map>(); + const metadata = { + get: async (t: string, n: string) => registry.get(t)?.get(n), + list: async (t: string) => [...(registry.get(t)?.values() ?? [])], + register: async (t: string, n: string, d: unknown) => { + if (!registry.has(t)) registry.set(t, new Map()); + registry.get(t)!.set(n, d); + }, + unregister: async (t: string, n: string) => { + registry.get(t)?.delete(n); + }, + listObjects: async () => [], + }; + + const logger = { warn: vi.fn(), info: vi.fn(), debug: vi.fn() }; + let service: DatasourceAdminService | undefined; + const ctx: any = { + getService: (name: string) => { + if (name === 'data') return engine; + if (name === 'metadata') return metadata; + throw new Error(`no service ${name}`); + }, + registerService: (name: string, svc: unknown) => { + if (name === 'datasource-admin') service = svc as DatasourceAdminService; + }, + trigger: async () => {}, + logger, + }; + + const factory = fakeFactory({ failWhen: opts.failWhen }); + const plugin = new DatasourceAdminServicePlugin({ driverFactory: factory, logger }); + await plugin.init(ctx); + await plugin.start(ctx); + + return { service: service!, drivers, defs, evicted, factory, registry, logger, plugin, ctx }; +} + +/** Both replicas over one store, joined (or not) by the bus. */ +async function makeCluster( + opts: { + attach?: boolean; + seed?: StoredDatasource[]; + peerFailWhen?: (spec: DatasourceConnectionSpec & { secret?: string }) => boolean; + } = {}, +) { + const store = makeSharedStore(); + for (const rec of opts.seed ?? []) store.rows.push(sysRow(rec)); + const writer = await makeReplica(store); + const peer = await makeReplica(store, { failWhen: opts.peerFailWhen }); + const { bus, raw, published, subscriptionCount } = makeBus(); + const detachers: Array<() => void> = []; + if (opts.attach !== false) { + detachers.push(writer.service.attachDatasourceMutationPubSub(bus, 'node-a')); + detachers.push(peer.service.attachDatasourceMutationPubSub(bus, 'node-b')); + } + return { store, writer, peer, bus, raw, published, subscriptionCount, detachers }; +} + +const settle = () => new Promise((r) => setTimeout(r, 0)); + +const draft = (over: Partial = {}): DatasourceDraft => ({ + name: 'analytics', + driver: 'sqlite', + config: { filename: '/tmp/old.db' }, + ...over, +}); + +const seeded: StoredDatasource = { + name: 'analytics', + driver: 'sqlite', + config: { filename: '/tmp/old.db' }, + origin: 'runtime', +}; + +describe('[#13805] ⭐ two-arm: the peer’s driver registry converges because of the bridge', () => { + it('Arm B — DELETE on the writer evicts the driver on the PEER, not only on the replica that served it', async () => { + const c = await makeCluster({ seed: [seeded] }); + // Both replicas rehydrated the pool at boot — the production shape. + const writerDriver = c.writer.drivers.get('analytics')!; + const peerDriver = c.peer.drivers.get('analytics')!; + expect(writerDriver.connected).toBe(true); + expect(peerDriver.connected).toBe(true); + + await c.writer.service.removeDatasource('analytics'); + + // The serving replica: #13578's eviction, unchanged. + expect(c.writer.drivers.has('analytics')).toBe(false); + // The card: the PEER evicts too — pre-fix this stayed `true` until restart. + await vi.waitFor(() => expect(c.peer.drivers.has('analytics')).toBe(false)); + expect(c.peer.evicted).toContain('analytics'); + expect(peerDriver.closed).toBe(true); + // The peer's def went with its driver (the registry owns that invariant). + expect(c.peer.defs.has('analytics')).toBe(false); + }); + + it('Arm A — CONTROL: the identical DELETE with no bridge leaves the peer’s driver in place', async () => { + const c = await makeCluster({ attach: false, seed: [seeded] }); + expect(c.writer.drivers.has('analytics')).toBe(true); + expect(c.peer.drivers.has('analytics')).toBe(true); + + await c.writer.service.removeDatasource('analytics'); + await settle(); + + expect(c.writer.drivers.has('analytics')).toBe(false); + // The pre-fix production shape: N-1 replicas keep the stuck driver. This + // arm is what makes Arm B a measurement — the harness shares a store + // between the replicas, never a registry. + expect(c.peer.drivers.has('analytics')).toBe(true); + expect(c.peer.drivers.get('analytics')!.closed).toBe(false); + expect(c.published).toHaveLength(0); + }); + + it('Arm B — create on the writer builds the pool on the PEER, from the peer’s OWN read of the shared row', async () => { + const c = await makeCluster(); + expect(c.peer.drivers.size).toBe(0); + + await c.writer.service.createDatasource(draft()); + + expect(c.writer.drivers.get('analytics')!.connected).toBe(true); + await vi.waitFor(() => expect(c.peer.drivers.has('analytics')).toBe(true)); + const peerDriver = c.peer.drivers.get('analytics')!; + expect(peerDriver.connected).toBe(true); + // Built from the SHARED ROW — the wire carried no config to build from. + expect(peerDriver.builtFrom.config).toMatchObject({ filename: '/tmp/old.db' }); + expect(c.peer.factory.created).toHaveLength(1); + // Converged the DRIVER registry only: the peer's metadata registry is a + // different sink (#13609's lane) and was deliberately left untouched. + expect(c.peer.registry.get('datasource')?.has('analytics') ?? false).toBe(false); + }); + + it('Arm A — CONTROL: the identical create with no bridge leaves the peer pool-less', async () => { + const c = await makeCluster({ attach: false }); + + await c.writer.service.createDatasource(draft()); + await settle(); + + expect(c.writer.drivers.has('analytics')).toBe(true); + expect(c.peer.drivers.size).toBe(0); + expect(c.peer.factory.created).toHaveLength(0); + }); + + it('Arm B — a connectivity change on the writer rebuilds the peer’s pool in place from the new row', async () => { + const c = await makeCluster({ seed: [seeded] }); + const oldPeerDriver = c.peer.drivers.get('analytics')!; + + await c.writer.service.updateDatasource('analytics', { config: { filename: '/tmp/new.db' } }); + + await vi.waitFor(() => expect(c.peer.drivers.get('analytics')).not.toBe(oldPeerDriver)); + const newPeerDriver = c.peer.drivers.get('analytics')!; + expect(newPeerDriver.builtFrom.config).toMatchObject({ filename: '/tmp/new.db' }); + expect(newPeerDriver.connected).toBe(true); + // The replaced pool's connection is closed, not leaked. + expect(oldPeerDriver.closed).toBe(true); + }); + + it('Arm B — active:false on the writer takes the peer out of service; active:true brings it back', async () => { + const c = await makeCluster({ seed: [seeded] }); + const peerDriver = c.peer.drivers.get('analytics')!; + + await c.writer.service.updateDatasource('analytics', { active: false }); + await vi.waitFor(() => expect(c.peer.drivers.has('analytics')).toBe(false)); + expect(peerDriver.closed).toBe(true); + + await c.writer.service.updateDatasource('analytics', { active: true }); + await vi.waitFor(() => expect(c.peer.drivers.has('analytics')).toBe(true)); + expect(c.peer.drivers.get('analytics')!.builtFrom.config).toMatchObject({ filename: '/tmp/old.db' }); + }); +}); + +describe('[#13805] the signal is an address, and receipt is convergence', () => { + it('publishes name + originNode only, on datasource.mutated, for create, update AND delete — symmetric', async () => { + const c = await makeCluster(); + + await c.writer.service.createDatasource(draft()); + await c.writer.service.updateDatasource('analytics', { label: 'Renamed' }); + await c.writer.service.removeDatasource('analytics'); + await settle(); + + expect(c.published).toHaveLength(3); + for (const { channel, payload } of c.published) { + expect(channel).toBe(DATASOURCE_MUTATION_CLUSTER_CHANNEL); + // Address-only: a peer must re-read its own store, so the wire must not + // offer it anything else to trust — no config, no driver, no verb. + expect(Object.keys(payload).sort()).toEqual(['name', 'originNode']); + expect(payload).toEqual({ originNode: 'node-a', name: 'analytics' }); + } + }); + + it('a duplicate delivery is a no-op — same driver instance on the peer, no rebuild', async () => { + const c = await makeCluster(); + await c.writer.service.createDatasource(draft()); + await vi.waitFor(() => expect(c.peer.drivers.has('analytics')).toBe(true)); + const peerDriver = c.peer.drivers.get('analytics')!; + const last = c.published[c.published.length - 1]; + + // What an at-least-once transport does: the same message, again. + await c.raw.publish(last.channel, last.payload); + await c.raw.publish(last.channel, last.payload); + await settle(); + + // Identity, not equivalence: a converge that rebuilt would pass an + // equivalence check with an identical pool. + expect(c.peer.drivers.get('analytics')).toBe(peerDriver); + expect(c.peer.factory.created).toHaveLength(1); + expect(c.peer.evicted).toHaveLength(0); + }); + + it('a label-only edit on the writer does not churn the peer’s pool', async () => { + const c = await makeCluster({ seed: [seeded] }); + const peerDriver = c.peer.drivers.get('analytics')!; + + await c.writer.service.updateDatasource('analytics', { label: 'Renamed' }); + await settle(); + + expect(c.published).toHaveLength(1); + expect(c.peer.drivers.get('analytics')).toBe(peerDriver); + expect(c.peer.factory.created).toHaveLength(1); + expect(c.peer.evicted).toHaveLength(0); + expect(peerDriver.closed).toBe(false); + }); + + it('two signals in quick succession converge in order — a create chased by an update ends on the update', async () => { + const c = await makeCluster(); + + // Not awaited in between: the peer sees both deliveries before its first + // convergence finishes reading. Ordered receipt is what keeps this from + // opening two pools for one name. + const create = c.writer.service.createDatasource(draft()); + const update = create.then(() => + c.writer.service.updateDatasource('analytics', { config: { filename: '/tmp/new.db' } }), + ); + await update; + + await vi.waitFor(() => + expect(c.peer.drivers.get('analytics')?.builtFrom.config).toMatchObject({ filename: '/tmp/new.db' }), + ); + // Every pool the peer opened along the way is either the live one or closed. + const stale = c.peer.factory.created.length - 1; + expect(stale).toBeGreaterThanOrEqual(0); + expect(c.peer.drivers.size).toBe(1); + }); + + it('a signal naming a datasource the peer never pooled, with no shared row, touches nothing — a code-defined pool survives a stray name', async () => { + const c = await makeCluster(); + // A pool the HOST STACK owns (code-defined): registered straight into the + // engine, never through this plugin's `registerPool`. + const codeDriver: FakeDriver = { + name: 'warehouse', + builtFrom: { name: 'warehouse', driver: 'postgres', config: {} }, + connected: true, + closed: false, + disconnect: async () => { + codeDriver.closed = true; + }, + }; + c.peer.drivers.set('warehouse', codeDriver); + + await c.raw.publish(DATASOURCE_MUTATION_CLUSTER_CHANNEL, { originNode: 'node-a', name: 'warehouse' }); + await c.raw.publish(DATASOURCE_MUTATION_CLUSTER_CHANNEL, { originNode: 'node-a', name: 'never_existed' }); + await settle(); + + expect(c.peer.drivers.get('warehouse')).toBe(codeDriver); + expect(codeDriver.closed).toBe(false); + expect(c.peer.evicted).toHaveLength(0); + expect(c.peer.factory.created).toHaveLength(0); + expect(c.peer.logger.warn).not.toHaveBeenCalled(); + }); + + it('a rebuild that fails on the peer keeps the peer’s OLD pool serving — never pool-less', async () => { + const c = await makeCluster({ + seed: [{ ...seeded, schemaMode: 'external', external: { allowWrites: false } }], + peerFailWhen: (spec) => (spec.config as { filename?: string }).filename === '/tmp/bad.db', + }); + const oldPeerDriver = c.peer.drivers.get('analytics')!; + + // The writer's own rebuild succeeds (its factory does not fail on the new + // config); only the peer's does not. + await c.writer.service.updateDatasource('analytics', { config: { filename: '/tmp/bad.db' } }); + await vi.waitFor(() => expect(c.peer.factory.created).toHaveLength(2)); + await settle(); + + // Same instance in the peer's registry, connection still open, def + // restored alongside — the #13804 keep-old-pool settlement, on a peer. + expect(c.peer.drivers.get('analytics')).toBe(oldPeerDriver); + expect(oldPeerDriver.closed).toBe(false); + expect(oldPeerDriver.connected).toBe(true); + expect(c.peer.defs.get('analytics')).toMatchObject({ name: 'analytics', schemaMode: 'external' }); + }); + + it('a stored row that cannot be read is NOT spent as "gone" — the peer keeps its pool and says so', async () => { + const c = await makeCluster({ seed: [seeded] }); + const peerDriver = c.peer.drivers.get('analytics')!; + const row = c.store.rows.find((r) => r.name === 'analytics')!; + row.metadata = '{not json'; + + await c.raw.publish(DATASOURCE_MUTATION_CLUSTER_CHANNEL, { originNode: 'node-a', name: 'analytics' }); + await settle(); + + expect(c.peer.drivers.get('analytics')).toBe(peerDriver); + expect(peerDriver.closed).toBe(false); + expect(c.peer.evicted).toHaveLength(0); + expect(c.peer.logger.warn).toHaveBeenCalledWith( + expect.stringContaining("converging 'analytics' after a peer write failed"), + expect.any(Error), + ); + }); + + it('a publish failure never fails the write it announces', async () => { + const c = await makeCluster(); + c.raw.publish = async () => { + throw new Error('bus down'); + }; + + const summary = await c.writer.service.createDatasource(draft()); + await settle(); + + expect(summary.name).toBe('analytics'); + expect(c.writer.drivers.get('analytics')!.connected).toBe(true); + expect(c.writer.logger.warn).toHaveBeenCalledWith( + expect.stringContaining("publishing 'analytics' to peer replicas failed"), + expect.any(Error), + ); + }); + + it('after detach, a write on the writer no longer reaches the peer', async () => { + const c = await makeCluster(); + for (const d of c.detachers) d(); + expect(c.subscriptionCount()).toBe(0); + + await c.writer.service.createDatasource(draft()); + await settle(); + + expect(c.published).toHaveLength(0); + expect(c.peer.drivers.size).toBe(0); + }); + + it('a replica that boots AFTER the write still converges the old way — boot rehydration is the loss bound', async () => { + const c = await makeCluster(); + await c.writer.service.createDatasource(draft()); + await settle(); + + // A third replica that was down at publish time: it never heard the + // signal, and does not need to — its boot reads the shared row. + const late = await makeReplica(c.store); + expect(late.drivers.get('analytics')?.connected).toBe(true); + }); +}); + +describe('[#13805] the attach seam — loopback, idempotency, the missing converge seam', () => { + function makeStub(over: Partial = {}) { + const logger = { warn: vi.fn(), info: vi.fn(), debug: vi.fn() }; + const config: DatasourceAdminServiceConfig = { + probe: async () => ({ ok: true }), + listDatasourceRecords: async () => [], + getDatasourceRecord: async () => undefined, + putDatasourceRecord: async () => {}, + deleteDatasourceRecord: async () => {}, + writeSecret: async () => 'ref', + countBoundObjects: async () => 0, + logger, + ...over, + }; + return { service: new DatasourceAdminService(config), logger }; + } + + it('ignores its own node’s messages (loopback) and converges on a peer’s', async () => { + const converge = vi.fn(async (_name: string) => {}); + const { service } = makeStub({ convergePool: converge }); + const { bus, raw } = makeBus(); + service.attachDatasourceMutationPubSub(bus, 'node-a'); + + await raw.publish(DATASOURCE_MUTATION_CLUSTER_CHANNEL, { originNode: 'node-a', name: 'analytics' }); + await settle(); + expect(converge).not.toHaveBeenCalled(); + + await raw.publish(DATASOURCE_MUTATION_CLUSTER_CHANNEL, { originNode: 'node-b', name: 'analytics' }); + await settle(); + expect(converge).toHaveBeenCalledTimes(1); + expect(converge).toHaveBeenCalledWith('analytics'); + }); + + it('a malformed payload is ignored, never converged', async () => { + const converge = vi.fn(async (_name: string) => {}); + const { service, logger } = makeStub({ convergePool: converge }); + const { bus, raw } = makeBus(); + service.attachDatasourceMutationPubSub(bus, 'node-a'); + + await raw.publish(DATASOURCE_MUTATION_CLUSTER_CHANNEL, { originNode: 'node-b' }); + await raw.publish(DATASOURCE_MUTATION_CLUSTER_CHANNEL, { originNode: 'node-b', name: '' }); + await raw.publish(DATASOURCE_MUTATION_CLUSTER_CHANNEL, null); + await settle(); + + expect(converge).not.toHaveBeenCalled(); + expect(logger.warn).not.toHaveBeenCalled(); + }); + + it('is idempotent on the (pubsub, nodeId) pair and detaches cleanly', async () => { + const { service, logger } = makeStub({ convergePool: async () => {} }); + const { bus, subscriptionCount } = makeBus(); + + const detach1 = service.attachDatasourceMutationPubSub(bus, 'node-a'); + const detach2 = service.attachDatasourceMutationPubSub(bus, 'node-a'); + expect(subscriptionCount()).toBe(1); + expect(logger.info).toHaveBeenCalledTimes(1); + expect(logger.info.mock.calls[0][0]).toBe( + `datasource admin: attached to the ${DATASOURCE_MUTATION_CLUSTER_CHANNEL} cluster channel (node=node-a)`, + ); + + // A different node id on the same transport is a re-attach. + service.attachDatasourceMutationPubSub(bus, 'node-a2'); + expect(subscriptionCount()).toBe(1); + + detach1(); + detach2(); + service.detachDatasourceMutationPubSub(); + expect(subscriptionCount()).toBe(0); + }); + + it('a host without a convergePool seam still publishes, applies nothing, and says so once at debug', async () => { + const { service, logger } = makeStub(); + const { bus, raw, published } = makeBus(); + service.attachDatasourceMutationPubSub(bus, 'node-a'); + + expect(logger.debug).toHaveBeenCalledTimes(1); + expect(logger.debug.mock.calls[0][0]).toContain('no convergePool seam wired'); + + await service.createDatasource(draft()); + expect(published).toEqual([ + { channel: DATASOURCE_MUTATION_CLUSTER_CHANNEL, payload: { originNode: 'node-a', name: 'analytics' } }, + ]); + + await raw.publish(DATASOURCE_MUTATION_CLUSTER_CHANNEL, { originNode: 'node-b', name: 'analytics' }); + await settle(); + expect(logger.warn).not.toHaveBeenCalled(); + }); + + it('migrateCredential does not publish — the live pool is left alone on every replica alike', async () => { + const records: StoredDatasource[] = [ + { + name: 'warehouse', + driver: 'postgres', + origin: 'runtime', + config: { host: 'db.internal', database: 'app', username: 'app', password: 'hunter2' }, + }, + ]; + const secrets = new Map(); + const { service } = makeStub({ + convergePool: async () => {}, + getDatasourceRecord: async (n) => records.find((r) => r.name === n), + putDatasourceRecord: async (r) => { + records.splice(0, records.length, r); + }, + writeSecret: async (input, hint) => { + const ref = `sys_secret://datasource/${hint.name}#1`; + secrets.set(ref, input.value); + return ref; + }, + readSecret: async (ref) => secrets.get(ref), + }); + const { bus, published } = makeBus(); + service.attachDatasourceMutationPubSub(bus, 'node-a'); + + const result = await service.migrateCredential('warehouse'); + + expect(result.status).toBe('migrated'); + expect(published).toHaveLength(0); + }); +}); diff --git a/packages/services/service-datasource/src/datasource-admin-plugin.ts b/packages/services/service-datasource/src/datasource-admin-plugin.ts index 2ad1c57545..19344b433d 100644 --- a/packages/services/service-datasource/src/datasource-admin-plugin.ts +++ b/packages/services/service-datasource/src/datasource-admin-plugin.ts @@ -17,6 +17,7 @@ import { DatasourceConnectionService, type ConnectionEngineLike, } from './datasource-connection-service.js'; +import { datasourceConnectivityChanged } from './datasource-connectivity-change.js'; import type { DatasourceConnectPolicy } from './contracts/connect-policy.js'; import type { Logger } from './logger.js'; import type { IDataEngine } from '@objectstack/spec/contracts'; @@ -145,6 +146,39 @@ async function loadDatasourceRows(engine: DataEngineLike | undefined): Promise { + if (!engine?.findOne) return undefined; + const row = await engine.findOne(SYS_METADATA, { where: { type: DS_META_TYPE, name, state: 'active' } }); + const raw = (row as { metadata?: unknown } | null | undefined)?.metadata; + if (raw == null) return undefined; + let parsed: Record; + try { + parsed = typeof raw === 'string' ? JSON.parse(raw) : (raw as Record); + } catch (err) { + throw new Error( + `datasource '${name}': the stored record could not be read (${err instanceof Error ? err.message : String(err)})`, + ); + } + // The same ADR-0087 rehydration pass `loadDatasourceRows` performs — this + // read bypasses the metadata service's converting loaders just as that one does. + return applyConversionsToStoredItem(DS_META_TYPE, parsed) as StoredDatasource; +} + /** * Host-provided secret binding. Encrypts a cleartext credential into the secret * store and returns an opaque `credentialsRef`; `unbind` deletes it. Wired by @@ -217,6 +251,18 @@ export class DatasourceAdminServicePlugin implements Plugin { private config?: DatasourceAdminServiceConfig; /** Shared "definition → live driver" path (ADR-0062 D1); also exposed as the `'datasource-connection'` service. */ private connection?: DatasourceConnectionService; + /** + * [#13805] The record each live runtime pool was last built from, by + * datasource name. Cluster convergence reads it twice: compared against the + * shared row it tells "unchanged" (a duplicate delivery, a label-only edit — + * the idempotent register, no churn) from "changed" (the in-place rebuild), + * and it rides along as `previous` so a failed rebuild restores the def the + * old pool was serving under — exactly what the serving replica's own + * update door does with the record it read. Written by the pool seams + * themselves, so boot rehydration and every admin door keep it current + * without knowing it exists; only pools THIS plugin built are ever in it. + */ + private readonly livePoolRecords = new Map(); private readonly options: DatasourceAdminServicePluginOptions; constructor(options: DatasourceAdminServicePluginOptions = {}) { @@ -403,12 +449,14 @@ export class DatasourceAdminServicePlugin implements Plugin { // Runtime-admin connects always degrade-with-warning on failure (never // fail-fast), preserving the pre-ADR-0062 admin behavior. registerPool: async (record) => { + this.livePoolRecords.set(record.name, record); await this.connection?.connect(record, { context: { origin: record.origin ?? 'runtime', trigger: 'runtime-admin' }, }); }, unregisterPool: async (name) => { + this.livePoolRecords.delete(name); await this.connection?.disconnect(name); }, @@ -419,12 +467,18 @@ export class DatasourceAdminServicePlugin implements Plugin { // datasource def with the driver, and the keep-old-pool path must put // back the def the old pool was serving under. reregisterPool: async (previous, next) => { + this.livePoolRecords.set(next.name, next); await this.connection?.reconnect(next, { previous, context: { origin: next.origin ?? 'runtime', trigger: 'runtime-admin' }, }); }, + // [#13805] The receive half of the datasource cluster bridge — see + // `convergePool` below for the decision table. The engine is resolved + // per call for the same reason every other seam here resolves lazily. + convergePool: (name) => this.convergePool(name, engineOf), + // The admin list's `status` reads the connection service's retained // verdicts (framework#3827). Resolved lazily per call: the service exists // by the end of this init(), but the verdicts only appear once boot @@ -554,6 +608,65 @@ export class DatasourceAdminServicePlugin implements Plugin { ); } + /** + * [#13805] Converge this replica's live pool for `name` from the SHARED + * record — the receive half of the datasource cluster bridge, run after a + * PEER's create, update or delete. It is boot's `restoreRuntimeDatasources` + * + `rehydratePools` reading, narrowed to one address and re-run on demand, + * against the durable `sys_metadata` row rather than this replica's + * metadata registry (which, on the host-config boot, is per-replica state + * with no cluster seam — the read a peer must not converge from): + * + * - **no active runtime row** (deleted, disabled, or a name this plugin + * never pooled) → evict the pool THIS plugin built, if any — through + * `unregisterPool` → `disconnect`, the #13578 eviction door. A name it + * never pooled is left alone, so a stray signal can never reach a + * code-defined pool the host stack owns. + * - **a row this replica has no pool for** → build it (the create case, and + * re-enabling after `active: false`). + * - **a row whose connectivity-bearing fields differ from what the live + * pool was built from** → rebuild in place, keeping the old pool on + * failure — `reregisterPool`, the same primitive and the same + * `datasourceConnectivityChanged` reading the serving replica's + * `updateDatasource` applies. + * - **a row that matches** → the idempotent register, a no-op while a + * driver holds the name (a duplicate delivery, a label-only edit) and a + * retry of a pool that failed to build (the serving replica's no-op-edit + * behaviour). + * + * ⛔ Deliberately does NOT write the metadata service's per-replica + * registry: which seam keeps `/api/v1/meta/datasource` coherent across + * replicas is a separate sink and a separate open measurement (#13609). + * This method converges the DRIVER registry only. + */ + private async convergePool( + name: string, + engineOf: () => DataEngineLike | undefined, + ): Promise { + const cfg = this.config; + if (!cfg?.registerPool || !cfg.unregisterPool) return; + + const row = await loadDatasourceRow(engineOf(), name); + const live = this.livePoolRecords.get(name); + + if (!row || row.origin !== 'runtime' || row.active === false) { + if (live) await cfg.unregisterPool(name); + return; + } + if (!live) { + await cfg.registerPool(row); + return; + } + if (datasourceConnectivityChanged(live, row)) { + // `reregisterPool` is always wired by this plugin; the fallback is the + // pre-#13804 idempotent register the admin service itself falls back to. + if (cfg.reregisterPool) await cfg.reregisterPool(live, row); + else await cfg.registerPool(row); + return; + } + await cfg.registerPool(row); + } + async destroy(): Promise { // Kernel teardown (ADR-0062 D5, #3993): close exactly the pools the shared // connection service opened — `'connected'` states only. `already-registered` diff --git a/packages/services/service-datasource/src/datasource-admin-service.ts b/packages/services/service-datasource/src/datasource-admin-service.ts index ba4305a958..bedd9a611a 100644 --- a/packages/services/service-datasource/src/datasource-admin-service.ts +++ b/packages/services/service-datasource/src/datasource-admin-service.ts @@ -31,6 +31,7 @@ */ import { validateDriverConfig } from '@objectstack/spec/data'; +import type { IPubSub } from '@objectstack/spec/contracts'; import { assertDatasourcePoolSupported } from './datasource-pool-support.js'; import { datasourceConnectivityChanged } from './datasource-connectivity-change.js'; import { redactDatasourceConfig, restoreRedactedConfig } from './datasource-config-redaction.js'; @@ -48,6 +49,47 @@ import type { Logger } from './logger.js'; /** Datasource name rule (mirrors `DatasourceSchema.name`). */ const NAME_RE = /^[a-z_][a-z0-9_]*$/; +/** + * [#13805] Cluster channel for datasource record writes — the DRIVER + * registry's adopter of the cluster-invalidation family `metadata.mutated` + * (#13331) established (ruled 2026-09-01: the same bridge shape, a symmetric + * create + delete signal, convergence by re-reading the shared datasource + * records, no second propagation mechanism). + * + * Published by the {@link DatasourceAdminService} that performed a create, + * update or delete (see {@link DatasourceAdminService.attachDatasourceMutationPubSub}) + * and consumed by every peer replica, which CONVERGES its own live pool for + * that name from its OWN read of the shared datasource records — it never + * replays the write. Distinct from `metadata.mutated` on purpose: that channel + * repairs the ObjectQL OBJECT registry behind `/api/v1/meta/*`, while this one + * repairs the ObjectQL DRIVER registry behind `/api/v1/datasources/*` — the + * state `registerPool` / `unregisterPool` mutate on the serving replica only, + * so that before this channel a `DELETE` recovered `/api/v1/ready` on the one + * replica that served it and every other replica kept the stuck driver until + * restart. + */ +export const DATASOURCE_MUTATION_CLUSTER_CHANNEL = 'datasource.mutated'; + +/** + * [#13805] Payload for {@link DATASOURCE_MUTATION_CLUSTER_CHANNEL}. + * + * ⭐ A SIGNAL, never trusted content — the contract `ClusterMetadataMutationPayload` + * carries, adopted verbatim: only the record's ADDRESS rides the channel. The + * receiving replica re-reads the shared record for `name` and converges its + * pool from that read — builds what the read says should be live, rebuilds + * what changed, evicts what is gone — so the shared store stays the single + * source of truth, and a duplicate or re-ordered delivery converges to the + * same pool state as a single one. Symmetric by construction: create, update + * and delete all publish this one shape, and a receiver cannot tell which it + * was without reading, which is the point. + */ +export interface ClusterDatasourceMutationPayload { + /** Origin nodeId — used for loopback suppression. */ + originNode?: string; + /** The datasource whose shared record was just written. */ + name: string; +} + /** * A persisted datasource record (subset of `Datasource`). `origin` distinguishes * code-defined from runtime; `external.credentialsRef` is the opaque secret @@ -151,6 +193,23 @@ export interface DatasourceAdminServiceConfig { * the safe direction for a host that cannot express a swap. */ reregisterPool?: (previous: StoredDatasource, next: StoredDatasource) => Promise | void; + /** + * [#13805] Converge THIS replica's live pool for `name` after a PEER wrote + * the shared datasource record — the receive half of + * {@link DatasourceAdminService.attachDatasourceMutationPubSub}. + * + * The host owns it because the read it must perform is the SHARED one: the + * durable row every replica writes (`sys_metadata` in the shipped plugin), + * not the metadata service's per-replica registry — on the host-config + * boot that registry has no cluster seam of its own, and is exactly the + * stale read a peer must not converge from. Contract for the host: re-read + * that record, then build what is missing, rebuild what changed, evict what + * is gone — never replay the mutation, and never touch a pool the host did + * not build itself. Optional: a host without it still PUBLISHES (peers that + * converge get the signal) and applies nothing it receives, stated once at + * `debug` when the transport attaches. + */ + convergePool?: (name: string) => Promise | void; /** * Last connect verdict per datasource, from `DatasourceConnectionService` * (framework#3827). Absent (a host without the connection service) means the @@ -230,6 +289,140 @@ export class DatasourceAdminService implements IDatasourceAdminService { return this.config.logger; } + // ── [#13805] the cluster half of every record write ──────────────────────── + // + // Before this seam the driver registry had no cluster propagation in either + // direction: each replica filled its registry at boot from the shared + // records and mutated it only for the writes IT served. This seam adopts + // the family `metadata.mutated` established — the state owner publishes an + // address, peers converge from their own read — and adopts it SYMMETRICALLY: + // a delete-only broadcast would have made delete more cluster-aware than + // create, a new asymmetry in the opposite direction rather than a repair. + + /** The cluster transport, when a bridge attached one. [#13805] */ + private clusterPubSub?: IPubSub; + /** This node's cluster id — stamps `originNode` for loopback suppression. */ + private clusterNodeId?: string; + /** Disposer for the cluster subscription, when attached. */ + private clusterUnsubscribe?: () => void; + /** + * One in-flight convergence per datasource name. Two signals for one name + * in quick succession (a create chased by an update) must converge in + * order: two concurrent builds would both open a pool and the registry + * would keep one and discard the other — an open connection nothing closes. + * The publisher's `partitionKey` keeps the order per name on the wire; this + * keeps it on receipt. + */ + private readonly convergeChains = new Map>(); + + /** + * [#13805] Attach a cluster pub/sub transport so this replica's datasource + * writes fan out on {@link DATASOURCE_MUTATION_CLUSTER_CHANNEL} and peer + * writes converge this replica's live pools. Mirrors + * `ObjectStackProtocolImplementation.attachMetadataMutationPubSub()` — + * including its idempotency on the `(pubsub, nodeId)` pair — and is called + * the same way: by `MetadataClusterBridgePlugin` in + * `@objectstack/service-cluster`, once per kernel boot at `kernel:ready`, + * after both services exist. Only `IPubSub` from `@objectstack/spec/contracts` + * crosses this seam: this package takes no dependency on the cluster + * service, and `@objectstack/objectql` — the registry's owner — is handed + * no bus at all (the layering line the card's triage drew). + * + * ⭐ Receipt runs CONVERGENCE, not trust: the payload is an address, and the + * host's `convergePool` re-reads the shared record before touching any + * pool. Duplicates are harmless (same read, same pool state); loss is + * bounded the way it always was — by the next boot's full rehydration — + * because no shipped driver exceeds at-most-once delivery (`IPubSub`'s own + * contract). This channel narrows "until every replica restarts" to "one + * network hop"; it does not promise more. + * + * @returns a disposer that detaches the bridge. + */ + attachDatasourceMutationPubSub(pubsub: IPubSub, nodeId: string): () => void { + if (this.clusterPubSub === pubsub && this.clusterNodeId === nodeId) { + return () => this.detachDatasourceMutationPubSub(); + } + this.detachDatasourceMutationPubSub(); + this.clusterPubSub = pubsub; + this.clusterNodeId = nodeId; + if (!this.config.convergePool) { + this.logger?.debug?.( + `datasource admin: no convergePool seam wired — peer writes on ${DATASOURCE_MUTATION_CLUSTER_CHANNEL} are published from here but not applied here`, + ); + } + this.clusterUnsubscribe = pubsub.subscribe( + DATASOURCE_MUTATION_CLUSTER_CHANNEL, + (msg) => { + const p = msg.payload; + // Loopback guard — never re-converge what this node just wrote: the + // pool seam already ran here, synchronously, at the door that + // persisted the record. + if (p?.originNode && p.originNode === this.clusterNodeId) return; + if (typeof p?.name !== 'string' || p.name.length === 0) return; + const converge = this.config.convergePool; + if (!converge) return; + this.enqueueConvergence(p.name, converge); + }, + ); + this.logger?.info?.( + `datasource admin: attached to the ${DATASOURCE_MUTATION_CLUSTER_CHANNEL} cluster channel (node=${nodeId})`, + ); + return () => this.detachDatasourceMutationPubSub(); + } + + /** Tear down the cluster wiring. Safe to call multiple times. [#13805] */ + detachDatasourceMutationPubSub(): void { + if (this.clusterUnsubscribe) { + try { + this.clusterUnsubscribe(); + } catch { + /* idempotent */ + } + this.clusterUnsubscribe = undefined; + } + this.clusterPubSub = undefined; + this.clusterNodeId = undefined; + } + + /** Serialise a peer signal's convergence behind any in flight for the same name. */ + private enqueueConvergence(name: string, converge: NonNullable): void { + const previous = this.convergeChains.get(name) ?? Promise.resolve(); + const next = previous + .then(() => converge(name)) + .catch((err) => { + this.logger?.warn(`datasource admin: converging '${name}' after a peer write failed`, err); + }); + this.convergeChains.set(name, next); + void next.then(() => { + if (this.convergeChains.get(name) === next) this.convergeChains.delete(name); + }); + } + + /** + * [#13805] The cluster half of a record write: publish the datasource's + * ADDRESS to peers. Best-effort and fire-and-forget — a publish failure must + * never fail the write it announces (the record is already persisted and + * this replica's pool already follows it), matching the protocol's + * `publishMetadataMutation` verbatim. No-op until a bridge attaches a + * transport. + * + * Called from the three doors that move a record: create, update, remove. + * Deliberately NOT from `migrateCredential`: that write leaves the live pool + * alone by design (the credential VALUE is unchanged, only where it is read + * from moves), so a peer has nothing to converge — and a signal would have + * every peer rebuild a working pool over a `credentialsRef` that + * dereferences to the same secret. + */ + private publishDatasourceMutation(name: string): void { + if (!this.clusterPubSub) return; + const payload: ClusterDatasourceMutationPayload = { originNode: this.clusterNodeId, name }; + void this.clusterPubSub + .publish(DATASOURCE_MUTATION_CLUSTER_CHANNEL, payload, { partitionKey: `datasource:${name}` }) + .catch((err) => { + this.logger?.warn(`datasource admin: publishing '${name}' to peer replicas failed`, err); + }); + } + async listDatasources(): Promise { const records = await this.config.listDatasourceRecords(); @@ -396,6 +589,9 @@ export class DatasourceAdminService implements IDatasourceAdminService { // `active ?? true` — so this door was the one place a deliberately // disabled datasource still came up serving. if (record.active !== false) await this.tryRegisterPool(record); + // [#13805] After the pool seam, whatever it decided: peers converge from + // the shared record, not from what this replica did with its own pool. + this.publishDatasourceMutation(record.name); return this.toSummary(record); } @@ -516,6 +712,10 @@ export class DatasourceAdminService implements IDatasourceAdminService { } else { await this.tryRegisterPool(merged); } + // [#13805] Unconditionally — a label-only edit included. The peer, not the + // signal, decides whether anything connectivity-bearing changed, by the + // same `datasourceConnectivityChanged` reading this door just applied. + this.publishDatasourceMutation(name); return this.toSummary(merged); } @@ -580,6 +780,10 @@ export class DatasourceAdminService implements IDatasourceAdminService { await this.config.deleteDatasourceRecord(name); if (existing.external?.credentialsRef) await this.tryRemoveSecret(existing.external.credentialsRef); await this.tryUnregisterPool(name); + // [#13805] The propagation half #13578 declared out of scope: the eviction + // above recovers THIS replica; the signal lets every other replica re-read + // the (now absent) shared record and evict its own copy of the driver. + this.publishDatasourceMutation(name); } /** diff --git a/packages/services/service-datasource/src/index.ts b/packages/services/service-datasource/src/index.ts index 00857c8a9d..c4bba8675c 100644 --- a/packages/services/service-datasource/src/index.ts +++ b/packages/services/service-datasource/src/index.ts @@ -75,6 +75,14 @@ export type { ProbeInput, } from './datasource-admin-service.js'; +// [#13805] The driver registry's cluster channel — the address-only signal a +// datasource write publishes so peer replicas converge their live pools from +// their own read of the shared record. Exported beside `metadata.mutated`'s +// constant for the same reason: a host wiring its own transport or subscriber +// names the channel it is on, rather than re-spelling it. +export { DATASOURCE_MUTATION_CLUSTER_CHANNEL } from './datasource-admin-service.js'; +export type { ClusterDatasourceMutationPayload } from './datasource-admin-service.js'; + // Which update actually changes what the live pool was built from (#13804) — // exported so a host wiring its own `reregisterPool` seam asks the same // question the shipped update path asks, rather than re-deriving the set. diff --git a/scripts/engine-double-contract.pinned.json b/scripts/engine-double-contract.pinned.json index 0dd09cc3ad..b3a9d5399f 100644 --- a/scripts/engine-double-contract.pinned.json +++ b/scripts/engine-double-contract.pinned.json @@ -3161,6 +3161,21 @@ "verb": "findOne", "pinned": 1 }, + { + "file": "packages/services/service-datasource/src/__tests__/datasource-cluster-convergence.test.ts", + "verb": "delete", + "pinned": 1 + }, + { + "file": "packages/services/service-datasource/src/__tests__/datasource-cluster-convergence.test.ts", + "verb": "findOne", + "pinned": 1 + }, + { + "file": "packages/services/service-datasource/src/__tests__/datasource-cluster-convergence.test.ts", + "verb": "update", + "pinned": 1 + }, { "file": "packages/services/service-job/src/db-job-adapter.degraded-outcome.test.ts", "verb": "update", From 6be669f67747210071d0d03fda58ef8b7921b537 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 2 Sep 2026 02:25:56 +0000 Subject: [PATCH 2/2] fix(service-datasource): cast the converted row through unknown; narrow the stray-signal warn assertion Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01AUF1NoViznQK32gqpK8wS8 --- .../src/__tests__/datasource-cluster-convergence.test.ts | 5 ++++- .../service-datasource/src/datasource-admin-plugin.ts | 2 +- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/packages/services/service-datasource/src/__tests__/datasource-cluster-convergence.test.ts b/packages/services/service-datasource/src/__tests__/datasource-cluster-convergence.test.ts index 71a85dc6a5..2afb6b6c2c 100644 --- a/packages/services/service-datasource/src/__tests__/datasource-cluster-convergence.test.ts +++ b/packages/services/service-datasource/src/__tests__/datasource-cluster-convergence.test.ts @@ -486,7 +486,10 @@ describe('[#13805] the signal is an address, and receipt is convergence', () => expect(codeDriver.closed).toBe(false); expect(c.peer.evicted).toHaveLength(0); expect(c.peer.factory.created).toHaveLength(0); - expect(c.peer.logger.warn).not.toHaveBeenCalled(); + // Nothing to converge is not a failure to converge: no warn from the + // receive path (the harness's boot-time "nav contribution skipped" warn + // is the plugin's own, unrelated to this seam). + expect(c.peer.logger.warn.mock.calls.some(([m]) => String(m).includes('converging'))).toBe(false); }); it('a rebuild that fails on the peer keeps the peer’s OLD pool serving — never pool-less', async () => { diff --git a/packages/services/service-datasource/src/datasource-admin-plugin.ts b/packages/services/service-datasource/src/datasource-admin-plugin.ts index 19344b433d..c118564f6e 100644 --- a/packages/services/service-datasource/src/datasource-admin-plugin.ts +++ b/packages/services/service-datasource/src/datasource-admin-plugin.ts @@ -176,7 +176,7 @@ async function loadDatasourceRow( } // The same ADR-0087 rehydration pass `loadDatasourceRows` performs — this // read bypasses the metadata service's converting loaders just as that one does. - return applyConversionsToStoredItem(DS_META_TYPE, parsed) as StoredDatasource; + return applyConversionsToStoredItem(DS_META_TYPE, parsed) as unknown as StoredDatasource; } /**