diff --git a/.changeset/27428-connector-credential-cluster-lane.md b/.changeset/27428-connector-credential-cluster-lane.md new file mode 100644 index 00000000000..23517b1db6f --- /dev/null +++ b/.changeset/27428-connector-credential-cluster-lane.md @@ -0,0 +1,16 @@ +--- +'@objectstack/service-automation': patch +'@objectstack/service-cluster': patch +--- + +fix(service-automation): a connector credential written or cleared through the door reaches every replica, not only the one that took the request + +Clause-②: no +Metadata-protocol: none + +`PUT` and `DELETE /api/v1/automation/connectors/credential-refs/:ref` reconciled the declarative connector instances of the replica that served the request. On a multi-replica deployment every other replica kept what it had: an instance whose stored value its upstream had refused is held with no retry timer, so a corrected credential took effect there only after a metadata reload or a restart, and requests landing on those replicas kept failing against a degraded connector. + +- **A fourth cluster lane.** After the row is written or cleared, the door publishes `{ originNode, ref }` on the `connector-credential.mutated` cluster channel. The payload is an address only: never a value, a fingerprint or a revision. +- **Peers converge from their own read.** A peer drops its own `originNode`, then runs its connector reconcile on its reconcile queue, which re-reads the shared `sys_connector_credential` rows. Exactly the instances naming a changed reference re-materialize, held ones included; a cleared reference degrades them, naming it. +- **Attached where the other lanes are.** `MetadataClusterBridgePlugin` (`@objectstack/service-cluster`) attaches the lane at `kernel:ready` beside `metadata.changed`, `metadata.mutated` and `datasource.mutated`, duck-typed against the automation service's `attachConnectorCredentialPubSub()`. As with lanes 2 and 3, it is skipped on the in-process `memory` cluster driver, so a single replica behaves as before. +- **What it does not promise.** Delivery is at-most-once, as for every lane of the family: a lost message is not retried, and the next metadata reload or restart still reconciles that replica. diff --git a/content/docs/kernel/cluster.mdx b/content/docs/kernel/cluster.mdx index 53e04ee981c..082611d0464 100644 --- a/content/docs/kernel/cluster.mdx +++ b/content/docs/kernel/cluster.mdx @@ -298,7 +298,7 @@ they declare scope and strategy, and the runtime does the rest. Metadata is the most cache-hot data in the system. Every read path — ObjectQL planning, REST routing, UI rendering, permission checks — touches it. Making it correct across a cluster needs a well-defined -invalidation signal per state owner — the three lanes of §6.2 — and, in this +invalidation signal per state owner — the four lanes of §6.2 — and, in this protocol, deliberately **no** per-item version on the wire (§6.1). ### 6.1 Monotonic version @@ -326,39 +326,41 @@ cache compare it on receipt. ### 6.2 The cluster-invalidation family Cross-node metadata invalidation is not one channel — it is a **family** of -three independent PubSub lanes, late-bound by `MetadataClusterBridgePlugin` +four independent PubSub lanes, late-bound by `MetadataClusterBridgePlugin` (`@objectstack/service-cluster`) at `kernel:ready`, one per state owner that -goes stale when metadata changes on a different node: +goes stale when a shared record changes on a different node: | Lane | Channel | Payload type | Package | Converges | |---|---|---|---|---| | 1 — metadata service cache | `metadata.changed` | `ClusterMetadataChangedPayload` | `@objectstack/metadata` | `MetadataManager`'s registry/list caches | | 2 — ObjectQL object registry | `metadata.mutated` (`METADATA_MUTATION_CLUSTER_CHANNEL`, #13331) | `ClusterMetadataMutationPayload` | `@objectstack/metadata-protocol` | ObjectQL's OBJECT registry, from a `sys_metadata` re-read | | 3 — ObjectQL driver registry | `datasource.mutated` (`DATASOURCE_MUTATION_CLUSTER_CHANNEL`, #13805) | `ClusterDatasourceMutationPayload` | `@objectstack/service-datasource` | ObjectQL's DRIVER registry, from a shared datasource-record re-read | +| 4 — declarative connector instances | `connector-credential.mutated` (`CONNECTOR_CREDENTIAL_MUTATION_CLUSTER_CHANNEL`, #27428) | `ClusterConnectorCredentialMutationPayload` (`{ originNode, ref }`) | `@objectstack/service-automation` | the automation service's materialized connectors, from a reconcile that re-reads the shared `sys_connector_credential` rows; published by the credential door (`PUT` / `DELETE /api/v1/automation/connectors/credential-refs/:ref`) after the row is written or cleared | -Lanes 2 and 3 share one contract, stated once here rather than in each +Lanes 2 to 4 share one contract, stated once here rather than in each channel's own doc comment: - **Address-only signal.** No message body ever rides these channels — the - payload names only what changed (a metadata address, or a datasource - name), never the changed value. + payload names only what changed (a metadata address, a datasource name, + or a connector credential reference), never the changed value. - **Receipt runs convergence, not trust.** A peer that receives a signal does not apply the payload as content; it re-reads its OWN copy of the source of - truth (`sys_metadata` for lane 2, the shared datasource record for lane 3) - and converges its local registry from that read. A duplicate or re-ordered + truth (`sys_metadata` for lane 2, the shared datasource record for lane 3, + the shared credential rows for lane 4) and converges its local registry + from that read. A duplicate or re-ordered delivery converges to the same state as a single one. - **`originNode` loopback suppression.** Every payload carries the publishing node's id; a peer that receives its own `originNode` drops the message — - the write-through or pool rebuild already ran, synchronously, at the door - that persisted the change. + the write-through, pool rebuild or connector reconcile already ran, + synchronously, at the door that persisted the change. - **At-most-once, bounded by boot rehydration.** No shipped `IPubSub` driver exceeds at-most-once delivery (§4.2), so a lost message is not retried. Each lane narrows the staleness window this always had — "until the next full reload on restart" — to "one network hop"; it does not promise more. -All three lanes skip their fan-out on the same condition — the in-process +All four lanes skip their fan-out on the same condition — the in-process `memory` cluster driver, described for lane 1 immediately below, applies -identically to lanes 2 and 3 (#14021). +identically to lanes 2 to 4 (#14021). #### Lane 1 — `metadata.changed` (the metadata service cache) diff --git a/packages/services/service-automation/src/connector-credential-cluster.test.ts b/packages/services/service-automation/src/connector-credential-cluster.test.ts new file mode 100644 index 00000000000..467a2a84f93 --- /dev/null +++ b/packages/services/service-automation/src/connector-credential-cluster.test.ts @@ -0,0 +1,303 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * [#27428] Cross-replica convergence of a connector credential written + * through the door — lane 4 of the cluster-invalidation family + * (`content/docs/kernel/cluster.mdx` §6.2), measured over the topology the + * card describes: two `AutomationServicePlugin` boots ("replicas") over ONE + * shared database, each with its OWN materialized connectors, 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). Each replica is attached through `attachConnectorCredentialPubSub()`, + * the seam `MetadataClusterBridgePlugin` calls at `kernel:ready`; that + * bridge's own lane-4 cases live in `@objectstack/service-cluster`. + * + * The defect: the door reconciled the replica that took the request. An + * instance whose STORED value its upstream refused is held — no retry timer, + * and its signature is recomputed only on `metadata:reloaded` or at restart — + * so a corrected credential reached every other replica only after a publish + * or a restart. + * + * Directions declared before running: + * • lane attached on both replicas: a PUT on A re-materializes B's held + * instance with the new value, and a DELETE on A degrades B's live one, + * with no reload on B -> GREEN + * • control, B not attached: the same PUT leaves B held — the pre-fix + * production shape, so the convergence above is the lane's doing and not + * a harness artifact sharing state between the replicas -> GREEN + * + * The values below are obvious dummies. + */ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { LiteKernel } from '@objectstack/core'; +import { ObjectQL } from '@objectstack/objectql'; +import { SqlDriver } from '@objectstack/driver-sql'; +import { SysSecret } from '@objectstack/platform-objects'; +import type { CryptoContext, CryptoHandle, ICryptoProvider, IPubSub, PubSubHandler } from '@objectstack/spec/contracts'; +import { + ConnectorCredentialRejectedError, + type Connector, + type ConnectorProviderContext, + type ConnectorProviderFactory, +} from '@objectstack/spec/integration'; + +import { AutomationServicePlugin } from './plugin.js'; +import type { AutomationEngine } from './engine.js'; +import { SysConnectorCredential } from './sys-connector-credential.object.js'; +import { SysFlowCredential } from './sys-flow-credential.object.js'; +import { CONNECTOR_CREDENTIAL_MUTATION_CLUSTER_CHANNEL } from './connector-credentials.js'; + +const REF = 'DUMMY_BILLING_TOKEN_27428'; +const DUMMY_VALUE = 'dummy-value-aaaa-0001'; +const DUMMY_VALUE_2 = 'dummy-value-bbbb-0002'; +const ADMIN = { userId: 'usr_admin_27428' } as never; + +const flush = () => new Promise((r) => setTimeout(r, 0)); + +/** Reversible stand-in cipher — the seam's behaviour is the engine's, not the cipher's. */ +function fakeCrypto(): ICryptoProvider { + let n = 0; + return { + async encrypt(plain: string, _ctx: CryptoContext): Promise { + n += 1; + return { id: `sec_${n}`, kmsKeyId: 'test', alg: 'test-rev', version: 1, ciphertext: [...plain].reverse().join('') }; + }, + async decrypt(handle: CryptoHandle): Promise { + return [...handle.ciphertext].reverse().join(''); + }, + async rotateKey(handle: CryptoHandle): Promise { + return { ...handle, version: handle.version + 1 }; + }, + digest: (plain: string) => `d:${plain.length}`, + keyedDigest: async (plain: string) => `k:${plain.length}`, + }; +} + +/** A provider-bound instance naming {@link REF}. */ +function billing() { + return { + name: 'billing', + label: 'billing', + type: 'api', + provider: 'fake', + providerConfig: {}, + auth: { type: 'bearer', credentialRef: REF }, + _packageId: 'com.example.billing', + }; +} + +/** One replica's upstream: it refuses at materialization (the mcp shape) while `reject` is set. */ +function fakeProvider() { + const state = { materialized: [] as ConnectorProviderContext[], reject: false }; + const tokenOf = (ctx: ConnectorProviderContext) => (ctx.auth?.type === 'bearer' ? ctx.auth.token : undefined); + const factory: ConnectorProviderFactory = (ctx) => { + state.materialized.push(ctx); + if (state.reject) { + throw new ConnectorCredentialRejectedError(`fake provider: connector '${ctx.name}' was refused by its server (HTTP 401)`, 401); + } + return { + def: { + name: ctx.name, + label: ctx.label, + type: 'api', + authentication: { type: 'none' }, + actions: [{ key: 'request', label: 'Request' }], + } as unknown as Connector, + handlers: { request: async () => ({ status: 200, ok: true, body: {} }) }, + }; + }; + return { factory, state, last: () => tokenOf(state.materialized[state.materialized.length - 1]) }; +} + +/** + * The shared bus: every publish reaches every subscriber on the channel, + * publisher included, one hop later — the at-most-once remote driver shape. + */ +function sharedBus() { + const handlers = new Map>>(); + const published: Array<{ channel: string; payload: unknown; partitionKey?: string }> = []; + const bus: IPubSub = { + async publish(channel, payload, opts) { + published.push({ channel, payload, partitionKey: opts?.partitionKey }); + for (const h of [...(handlers.get(channel) ?? [])]) { + setTimeout(() => void h({ channel, payload, publishedAt: Date.now() }), 0); + } + }, + subscribe(channel, handler) { + const set = handlers.get(channel) ?? new Set(); + set.add(handler as PubSubHandler); + handlers.set(channel, set); + return () => set.delete(handler as PubSubHandler); + }, + async close() {}, + }; + return { bus, published }; +} + +const engines: ObjectQL[] = []; +const kernels: LiteKernel[] = []; +afterEach(async () => { + for (const k of kernels.splice(0)) await k.shutdown().catch(() => {}); + for (const e of engines.splice(0)) { + try { await (e as any).destroy?.(); } catch { /* noop */ } + } + vi.restoreAllMocks(); +}); + +/** Everything the process prints while a test runs (the kernel logger writes JSON to stdout / stderr). */ +function captureOutput(): { text: () => string } { + const chunks: string[] = []; + const keep = ((c: string | Uint8Array) => { chunks.push(String(c)); return true; }) as never; + vi.spyOn(process.stdout, 'write').mockImplementation(keep); + vi.spyOn(process.stderr, 'write').mockImplementation(keep); + for (const m of ['log', 'info', 'warn', 'error', 'debug'] as const) { + vi.spyOn(console, m).mockImplementation((...args: unknown[]) => { chunks.push(args.map(String).join(' ')); }); + } + return { text: () => chunks.join('\n') }; +} + +/** The ONE database both replicas read and write. */ +async function sharedDatabase(): Promise { + const ql = new ObjectQL(); + engines.push(ql); + const driver = new SqlDriver({ client: 'better-sqlite3', connection: { filename: ':memory:' }, useNullAsDefault: true }); + await driver.connect(); + ql.registerDriver(driver, true); + await ql.init(); + (ql as any).registry.registerObject(SysSecret as never, 'test'); + (ql as any).registry.registerObject(SysConnectorCredential as never, 'test'); + (ql as any).registry.registerObject(SysFlowCredential as never, 'test'); + await ql.syncSchemas(); + ql.setCryptoProvider(fakeCrypto()); + return ql; +} + +/** One replica: its own kernel, plugin and provider over the shared database; attached to `bus` unless told not to. */ +async function replica(ql: ObjectQL, nodeId: string, bus: IPubSub | undefined) { + const provider = fakeProvider(); + const kernel = new LiteKernel({ logger: { level: 'debug', format: 'json' } } as never); + kernels.push(kernel); + kernel.use(new AutomationServicePlugin({ suspendedRunStore: 'memory', credentialResolver: () => undefined })); + kernel.use({ + name: `test.harness.${nodeId}`, + type: 'standard' as const, + version: '1.0.0', + dependencies: ['com.objectstack.service-automation'], + async init(ctx: any) { + ctx.registerService('objectql', { registry: { listItems: (t: string) => (t === 'connector' ? [billing()] : []) } }); + ctx.registerService('data', ql); + ctx.getService('automation').registerConnectorProvider('fake', provider.factory); + }, + async start() {}, + } as never); + await kernel.bootstrap(); + await flush(); + const engine = kernel.getService('automation') as AutomationEngine; + // What `MetadataClusterBridgePlugin`'s lane 4 does at `kernel:ready` on a cross-process driver. + if (bus) engine.attachConnectorCredentialPubSub(bus, nodeId); + const held = () => engine.getConnectorDegradedReason('billing'); + return { engine, provider, held }; +} + +describe('[#27428] a credential written on one replica converges every replica', () => { + it('⭐ a PUT on A re-materializes the instance B holds, with the new value and no reload on B', async () => { + const ql = await sharedDatabase(); + const { bus } = sharedBus(); + const a = await replica(ql, 'node-a', bus); + const b = await replica(ql, 'node-b', bus); + a.provider.state.reject = true; + b.provider.state.reject = true; + + // The first value is refused by the upstream on both replicas: each + // holds its instance (a stored value — no retry timer). + await a.engine.setConnectorCredential({ ref: REF, value: DUMMY_VALUE, context: ADMIN }); + await vi.waitFor(() => expect(b.provider.state.materialized).toHaveLength(1)); + await vi.waitFor(() => expect(b.held()).toContain('refused')); + expect(b.provider.last()).toBe(DUMMY_VALUE); + expect(a.held()).toContain('refused'); + + // The administrator corrects it on A. B, which holds its instance, + // re-materializes with the new value — no `metadata:reloaded`, no restart. + a.provider.state.reject = false; + b.provider.state.reject = false; + await a.engine.setConnectorCredential({ ref: REF, value: DUMMY_VALUE_2, context: ADMIN }); + expect(a.held()).toBeUndefined(); + await vi.waitFor(() => expect(b.held()).toBeUndefined()); + expect(b.provider.last()).toBe(DUMMY_VALUE_2); + expect(b.engine.getRegisteredConnectors()).toContain('billing'); + }); + + it('a DELETE on A degrades the live instance on B, naming the reference', async () => { + const ql = await sharedDatabase(); + const { bus } = sharedBus(); + const a = await replica(ql, 'node-a', bus); + const b = await replica(ql, 'node-b', bus); + + await a.engine.setConnectorCredential({ ref: REF, value: DUMMY_VALUE, context: ADMIN }); + await vi.waitFor(() => expect(b.held()).toBeUndefined()); + expect(b.provider.last()).toBe(DUMMY_VALUE); + + await a.engine.clearConnectorCredential({ ref: REF, context: ADMIN }); + await vi.waitFor(() => expect(b.held()).toContain(`credential '${REF}'`)); + expect(a.held()).toContain(`credential '${REF}'`); + }); + + it('control: a replica with no transport attached keeps holding its instance — the pre-fix shape', async () => { + const ql = await sharedDatabase(); + const { bus } = sharedBus(); + const a = await replica(ql, 'node-a', bus); + const b = await replica(ql, 'node-b', undefined); + a.provider.state.reject = true; + b.provider.state.reject = true; + await b.engine.setConnectorCredential({ ref: REF, value: DUMMY_VALUE, context: ADMIN }); + expect(b.held()).toContain('refused'); + + a.provider.state.reject = false; + b.provider.state.reject = false; + await a.engine.setConnectorCredential({ ref: REF, value: DUMMY_VALUE_2, context: ADMIN }); + expect(a.held()).toBeUndefined(); + for (let i = 0; i < 5; i++) await flush(); + expect(b.held()).toContain('refused'); + expect(b.provider.state.materialized).toHaveLength(1); + }); +}); + +describe('[#27428] the lane carries an address, and the publisher drops its own', () => { + it('publishes `{ originNode, ref }` after a set and after a clear, never a value; a refused write publishes nothing', async () => { + const ql = await sharedDatabase(); + const { bus, published } = sharedBus(); + const a = await replica(ql, 'node-a', bus); + + await expect(a.engine.setConnectorCredential({ ref: REF, value: '', context: ADMIN })) + .rejects.toMatchObject({ status: 400, code: 'VALIDATION_ERROR' }); + await expect(a.engine.clearConnectorCredential({ ref: REF, context: ADMIN })) + .rejects.toMatchObject({ status: 404, code: 'RESOURCE_NOT_FOUND' }); + expect(published).toHaveLength(0); + + await a.engine.setConnectorCredential({ ref: REF, value: DUMMY_VALUE, context: ADMIN }); + await a.engine.clearConnectorCredential({ ref: REF, context: ADMIN }); + expect(published).toEqual([ + { channel: CONNECTOR_CREDENTIAL_MUTATION_CLUSTER_CHANNEL, payload: { originNode: 'node-a', ref: REF }, partitionKey: `connector-credential:${REF}` }, + { channel: CONNECTOR_CREDENTIAL_MUTATION_CLUSTER_CHANNEL, payload: { originNode: 'node-a', ref: REF }, partitionKey: `connector-credential:${REF}` }, + ]); + expect(JSON.stringify(published).includes(DUMMY_VALUE)).toBe(false); + }); + + it("A's own publish is dropped on A and followed on B; the lane runs in both directions", async () => { + const out = captureOutput(); + const ql = await sharedDatabase(); + const { bus } = sharedBus(); + const a = await replica(ql, 'node-a', bus); + const b = await replica(ql, 'node-b', bus); + + await a.engine.setConnectorCredential({ ref: REF, value: DUMMY_VALUE, context: ADMIN }); + await vi.waitFor(() => expect(out.text()).toContain(`reconciling the connectors on node 'node-b'`)); + for (let i = 0; i < 5; i++) await flush(); + expect(out.text()).not.toContain(`reconciling the connectors on node 'node-a'`); + + // B writes; A follows. + await b.engine.setConnectorCredential({ ref: REF, value: DUMMY_VALUE_2, context: ADMIN }); + await vi.waitFor(() => expect(a.provider.last()).toBe(DUMMY_VALUE_2)); + expect(out.text()).toContain(`connector credential '${REF}' was written on node 'node-b'; reconciling the connectors on node 'node-a'`); + }); +}); diff --git a/packages/services/service-automation/src/connector-credentials.ts b/packages/services/service-automation/src/connector-credentials.ts index f536fcd8706..3a05821ddca 100644 --- a/packages/services/service-automation/src/connector-credentials.ts +++ b/packages/services/service-automation/src/connector-credentials.ts @@ -41,6 +41,38 @@ export const CONNECTOR_CREDENTIAL_REF_MAX_LENGTH = 255; /** The bound on a value the door accepts. A PEM key or a service-account JSON fits several times over. */ export const CONNECTOR_CREDENTIAL_VALUE_MAX_LENGTH = 16_384; +/** + * [#27428] The cluster channel a credential door write fans out on — lane 4 + * of the cluster-invalidation family (`content/docs/kernel/cluster.mdx` + * §6.2), attached by `MetadataClusterBridgePlugin` in + * `@objectstack/service-cluster` beside `metadata.changed`, + * `metadata.mutated` and `datasource.mutated`. + * + * The door reconciles the replica that took the request; every other replica + * holds its own materialized instances, and one whose stored value its + * upstream refused is held with no retry timer — so before this channel a + * corrected credential took effect on one replica at once and on the others + * only after a metadata reload or a restart. + */ +export const CONNECTOR_CREDENTIAL_MUTATION_CLUSTER_CHANNEL = 'connector-credential.mutated'; + +/** + * [#27428] Payload for {@link CONNECTOR_CREDENTIAL_MUTATION_CLUSTER_CHANNEL}. + * + * ⭐ A SIGNAL, never trusted content — the family's contract: only the + * reference rides the channel, never a value, a fingerprint or a revision. + * The receiving replica runs its own reconcile, which re-reads the shared + * `sys_connector_credential` rows (revision, then fingerprint), so a + * duplicate or re-ordered delivery converges to the same instances as a + * single one. Symmetric: a set and a clear publish this one shape. + */ +export interface ClusterConnectorCredentialMutationPayload { + /** Origin nodeId — used for loopback suppression. */ + originNode?: string; + /** The reference whose stored row was just written or cleared. */ + ref: string; +} + /** Where a reference resolved from — never what it resolved to. */ export type ConnectorCredentialSource = 'env' | 'object' | 'vault'; diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 421cbefbbec..25b717fc8b4 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -20,7 +20,7 @@ import { declaredScreenFieldNames, type ScreenFieldVisibility, } from './screen-input-contract.js'; -import type { Logger } from '@objectstack/spec/contracts'; +import type { IPubSub, Logger } from '@objectstack/spec/contracts'; import type { FlowUnboundCause } from '@objectstack/spec/contracts'; import { FLOW_RESERVED_NODE_TYPES, FLOW_STRUCTURAL_NODE_TYPES } from '@objectstack/spec/automation'; // [#14328] The ONE answer to "which trigger kind does this flow ask for?" — @@ -2533,6 +2533,12 @@ export interface ConnectorCredentialDoor { list(): Promise; set(request: ConnectorCredentialSetRequest): Promise; clear(request: ConnectorCredentialClearRequest): Promise; + /** + * [#27428] Attach the cluster transport the door's writes fan out on, and + * whose peer writes reconcile this replica — see + * {@link AutomationEngine.attachConnectorCredentialPubSub}. + */ + attachClusterPubSub(pubsub: IPubSub, nodeId: string): () => void; } /** @@ -4800,6 +4806,21 @@ export class AutomationEngine implements IAutomationService { ); } + /** + * [#27428] Attach a cluster pub/sub transport to the connector credential + * door: a write through it publishes the reference on + * `connector-credential.mutated`, and a peer's write reconciles the + * instances held here. Called by `MetadataClusterBridgePlugin` in + * `@objectstack/service-cluster` at `kernel:ready` — lane 4 of + * `content/docs/kernel/cluster.mdx` §6.2, duck-typed there so neither + * package depends on the other. Only `IPubSub` crosses this seam. + * + * @returns a disposer that detaches the transport. + */ + attachConnectorCredentialPubSub(pubsub: IPubSub, nodeId: string): () => void { + return this.requireConnectorCredentialDoor('attachConnectorCredentialPubSub()').attachClusterPubSub(pubsub, nodeId); + } + /** [#27237] `IAutomationService.listConnectorCredentialRefs` — references and states, never a value. */ async listConnectorCredentialRefs(): Promise { return this.requireConnectorCredentialDoor('listConnectorCredentialRefs()').list(); diff --git a/packages/services/service-automation/src/plugin.ts b/packages/services/service-automation/src/plugin.ts index 8744db630ec..11ce269dc41 100644 --- a/packages/services/service-automation/src/plugin.ts +++ b/packages/services/service-automation/src/plugin.ts @@ -2,7 +2,7 @@ import type { Plugin, PluginContext } from '@objectstack/core'; import { effectiveTenancyPosture, isServiceNotRegisteredError, resolveLocalizationContext, resolveUserAuthzGrants } from '@objectstack/core'; -import type { IJobService } from '@objectstack/spec/contracts'; +import type { IJobService, IPubSub } from '@objectstack/spec/contracts'; import type { FlowFunctionEffect } from '@objectstack/spec/automation'; import type { Connector, @@ -37,6 +37,7 @@ import { CONNECTOR_CREDENTIAL_VALUE_MAX_LENGTH, ConnectorCredentialStore, ConnectorCredentialUnavailableError, + CONNECTOR_CREDENTIAL_MUTATION_CLUSTER_CHANNEL, SECRET_HANDLE_PREFIX, connectorCredentialFingerprint, credentialDoorRefusal, @@ -50,6 +51,7 @@ import { resolveConnectorCredential, upstreamRefusalMessage, withoutCredentialThrown, + type ClusterConnectorCredentialMutationPayload, type ConnectorCredentialChain, type ConnectorCredentialRecord, type ConnectorCredentialResolution, @@ -803,6 +805,12 @@ export class AutomationServicePlugin implements Plugin { * credential door writes. See `./connector-credentials.ts`. */ private connectorCredentialStore?: ConnectorCredentialStore; + /** + * [#27428] The cluster transport the credential door's writes fan out on, + * this node's id (stamps `originNode`), and the subscription's disposer — + * set by {@link attachCredentialClusterPubSub}. + */ + private credentialCluster?: { pubsub: IPubSub; nodeId: string; unsubscribe: () => void }; /** * [ADR-0015 D7] Moved by every reconcile — which every credential write * runs — so a datasource's borrowed credential is resolved again after a @@ -1081,6 +1089,7 @@ export class AutomationServicePlugin implements Plugin { list: () => this.listConnectorCredentialRefs(ctx), set: (request) => this.setConnectorCredential(ctx, request), clear: (request) => this.clearConnectorCredential(ctx, request), + attachClusterPubSub: (pubsub, nodeId) => this.attachCredentialClusterPubSub(ctx, pubsub, nodeId), }); // [ADR-0015 D7] The connector credential source: a datasource that @@ -2834,6 +2843,7 @@ export class AutomationServicePlugin implements Plugin { `[Automation] connector credential '${ref}' set (revision ${revision})${userId ? ` by user '${userId}'` : ''}; ` + `reconciling ${declaredBy.map((c) => `'${c.name}'`).join(', ')}.`, ); + this.publishCredentialMutation(ctx, ref); await this.reconcileDeclaredConnectors(ctx, { fatal: false }); }); return this.describeCredentialRef(chain, ref, declaredBy, await this.readCredentialRowsForDoor(chain)); @@ -2866,12 +2876,95 @@ export class AutomationServicePlugin implements Plugin { throw credentialDoorRefusal(`Nothing is stored for credential '${ref}'.`, 'RESOURCE_NOT_FOUND', 404); } ctx.logger.info(`[Automation] connector credential '${ref}' cleared${userId ? ` by user '${userId}'` : ''}.`); + this.publishCredentialMutation(ctx, ref); await this.reconcileDeclaredConnectors(ctx, { fatal: false }); }); const declared = await this.declaredCredentialRefs(ctx); return this.describeCredentialRef(chain, ref, declared.get(ref), await this.readCredentialRowsForDoor(chain)); } + /** + * [#27428] Attach the cluster transport for connector credential writes — + * the door half of lane 4 of the cluster-invalidation family + * (`content/docs/kernel/cluster.mdx` §6.2), reached through + * `AutomationEngine.attachConnectorCredentialPubSub()`, which + * `MetadataClusterBridgePlugin` calls once per boot at `kernel:ready`. + * Idempotent on the `(pubsub, nodeId)` pair, as lanes 2 and 3 are. + * + * ⭐ Receipt runs CONVERGENCE, not trust: a peer's signal names a + * reference, and this replica answers it with its own reconcile on the + * reconcile queue, which re-reads the shared rows — the revision joins + * each instance's signature, so exactly the instances naming a changed + * reference re-materialize, a held one included. A duplicate runs a + * reconcile that changes nothing. Loss is bounded the way it always was, + * by the next metadata reload or boot: no shipped driver exceeds + * at-most-once delivery. + * + * @returns a disposer that detaches the transport. + */ + private attachCredentialClusterPubSub(ctx: PluginContext, pubsub: IPubSub, nodeId: string): () => void { + const current = this.credentialCluster; + if (current?.pubsub === pubsub && current.nodeId === nodeId) { + return () => this.detachCredentialClusterPubSub(); + } + this.detachCredentialClusterPubSub(); + const unsubscribe = pubsub.subscribe( + CONNECTOR_CREDENTIAL_MUTATION_CLUSTER_CHANNEL, + (msg) => { + const p = msg.payload; + // Loopback guard — the door that wrote the row already + // reconciled this replica, on the same queue. + if (p?.originNode && p.originNode === nodeId) return; + if (typeof p?.ref !== 'string' || p.ref.length === 0) return; + ctx.logger.info( + `[Automation] connector credential '${p.ref}' was written on node '${p.originNode ?? 'unknown'}'; ` + + `reconciling the connectors on node '${nodeId}'.`, + ); + void this.materializeDeclaredConnectors(ctx, { fatal: false }).catch((err) => { + ctx.logger.warn( + `[Automation] reconciling after a peer's write of connector credential '${p.ref}' failed; ` + + 'the next metadata reload or restart reconciles it.', + describeThrownForLog(err), + ); + }); + }, + ); + this.credentialCluster = { pubsub, nodeId, unsubscribe }; + return () => this.detachCredentialClusterPubSub(); + } + + /** [#27428] Tear down the credential cluster wiring. Safe to call multiple times. */ + private detachCredentialClusterPubSub(): void { + const current = this.credentialCluster; + this.credentialCluster = undefined; + try { + current?.unsubscribe(); + } catch { + /* idempotent */ + } + } + + /** + * [#27428] The cluster half of a credential door write: publish the + * reference to peers, after the row is written or cleared. Best-effort and + * fire-and-forget, as lanes 2 and 3 publish — the row is already stored + * and this replica's reconcile follows it, so a publish failure never + * fails the write it announces. No-op until a bridge attaches a transport. + */ + private publishCredentialMutation(ctx: PluginContext, ref: string): void { + const cluster = this.credentialCluster; + if (!cluster) return; + const payload: ClusterConnectorCredentialMutationPayload = { originNode: cluster.nodeId, ref }; + void cluster.pubsub + .publish(CONNECTOR_CREDENTIAL_MUTATION_CLUSTER_CHANNEL, payload, { partitionKey: `connector-credential:${ref}` }) + .catch((err) => { + ctx.logger.warn( + `[Automation] publishing connector credential '${ref}' to peer replicas failed; they follow it on their next metadata reload or restart.`, + describeThrownForLog(err), + ); + }); + } + /** * [#27237] Degrade one instance whose upstream refused its credential — * at materialization (`ConnectorCredentialRejectedError`, the mcp @@ -3602,6 +3695,8 @@ export class AutomationServicePlugin implements Plugin { // Stop the degraded-instance retry loop first (#3017): mark destroyed so // an already-queued reconcile no-ops, and cancel any armed timer. this.destroyed = true; + // [#27428] No peer credential write reconciles after shutdown. + this.detachCredentialClusterPubSub(); // [#20790] No credential move after shutdown, and none left in flight. this.unsubscribeCryptoProvider?.(); this.unsubscribeCryptoProvider = undefined; 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 fc4cfbcb232..a5f872ea2ca 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 @@ -57,12 +57,22 @@ interface HarnessOptions { datasourceAdmin?: 'none' | 'bare' | 'real'; /** When true, the datasource-admin seam throws on attach. */ datasourceAttachThrows?: boolean; + /** + * [#27428] The `automation` slot: `'none'` (getService throws — the + * default, so every earlier case reads exactly as it did), `'bare'` + * (present, no attachConnectorCredentialPubSub — an older implementation), + * or `'real'` (exposes attachConnectorCredentialPubSub). + */ + automation?: 'none' | 'bare' | 'real'; + /** When true, the automation seam throws on attach. */ + automationAttachThrows?: boolean; } function makeHarness(opts: HarnessOptions = {}) { const { driver = 'redis', metadata = 'fallback', protocol = 'real', datasourceAdmin = 'none', datasourceAttachThrows = false, + automation = 'none', automationAttachThrows = false, } = opts; const logger = { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }; @@ -76,6 +86,11 @@ function makeHarness(opts: HarnessOptions = {}) { if (datasourceAttachThrows) throw new Error('datasource attach exploded'); return detachDatasource; }); + const detachCredential = vi.fn(); + const attachCredential = vi.fn((_pubsub: unknown, _nodeId: string) => { + if (automationAttachThrows) throw new Error('automation attach exploded'); + return detachCredential; + }); const pubsub = { publish: vi.fn(), subscribe: vi.fn(), close: vi.fn() }; const cluster = @@ -95,6 +110,10 @@ function makeHarness(opts: HarnessOptions = {}) { datasourceAdmin === 'none' ? undefined : datasourceAdmin === 'bare' ? { listDatasources: vi.fn() } : { attachDatasourceMutationPubSub: attachDatasource }; + const automationService = + automation === 'none' ? undefined + : automation === 'bare' ? { execute: vi.fn() } + : { attachConnectorCredentialPubSub: attachCredential }; const hooks = new Map Promise | void>>(); const ctx = { @@ -121,6 +140,10 @@ function makeHarness(opts: HarnessOptions = {}) { if (!datasourceAdminService) throw new Error('service not found: datasource-admin'); return datasourceAdminService; } + if (name === 'automation') { + if (!automationService) throw new Error('service not found: automation'); + return automationService; + } throw new Error(`service not found: ${name}`); }, } as unknown as PluginContext; @@ -133,6 +156,7 @@ function makeHarness(opts: HarnessOptions = {}) { ctx, logger, fire, pubsub, attachMetadata, detachMetadata, attachMutation, detachMutation, attachDatasource, detachDatasource, + attachCredential, detachCredential, }; } @@ -412,3 +436,85 @@ describe('[#13805] lane 3 — the datasource admin service’s datasource.mutate expect(h.detachDatasource).toHaveBeenCalledTimes(1); }); }); + +describe('[#27428] lane 4 — the automation service’s connector-credential.mutated fan-out', () => { + it('⭐ attaches on a cross-process driver with the cluster’s pubsub and node id, and reports it — independently of lanes 1 to 3', async () => { + const h = makeHarness({ driver: 'redis', metadata: 'none', protocol: 'none', automation: 'real' }); + await new MetadataClusterBridgePlugin().init(h.ctx); + await h.fire('kernel:ready'); + + expect(h.attachCredential).toHaveBeenCalledTimes(1); + expect(h.attachCredential).toHaveBeenCalledWith(h.pubsub, 'node-a'); + expect(infoLines(h)).toContain( + 'MetadataClusterBridgePlugin: bridged connector-credential.mutated → cluster.pubsub (node=node-a)', + ); + }); + + it('all four lanes attach together when every owner exposes its seam', async () => { + const h = makeHarness({ + driver: 'redis', metadata: 'manager', protocol: 'real', datasourceAdmin: 'real', automation: 'real', + }); + await new MetadataClusterBridgePlugin().init(h.ctx); + await h.fire('kernel:ready'); + + expect(h.attachMetadata).toHaveBeenCalledTimes(1); + expect(h.attachMutation).toHaveBeenCalledTimes(1); + expect(h.attachDatasource).toHaveBeenCalledTimes(1); + expect(h.attachCredential).toHaveBeenCalledTimes(1); + }); + + it('skips attach on the in-process memory driver — no peers to reach, nothing said above debug', async () => { + const h = makeHarness({ driver: 'memory', automation: 'real' }); + await new MetadataClusterBridgePlugin().init(h.ctx); + await h.fire('kernel:ready'); + + expect(h.attachCredential).not.toHaveBeenCalled(); + expect(infoLines(h).some((l) => l.includes('connector-credential.mutated'))).toBe(false); + expect( + debugLines(h).some((l) => l.includes('is in-process') && l.includes('connector-credential fan-out')), + ).toBe(true); + }); + + it('skips quietly when no automation service is registered, or it does not expose the seam', async () => { + for (const automation of ['none', 'bare'] as const) { + const h = makeHarness({ driver: 'redis', automation }); + await new MetadataClusterBridgePlugin().init(h.ctx); + await h.fire('kernel:ready'); + + expect(h.attachCredential).not.toHaveBeenCalled(); + expect(h.logger.error).not.toHaveBeenCalled(); + expect(warnLines(h).some((l) => l.includes('connector-credential'))).toBe(false); + } + }); + + 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', + automation: 'real', automationAttachThrows: true, + }); + await new MetadataClusterBridgePlugin().init(h.ctx); + await h.fire('kernel:ready'); + + expect(h.attachDatasource).toHaveBeenCalledTimes(1); + expect(h.logger.error).toHaveBeenCalledWith( + 'MetadataClusterBridgePlugin: connector-credential-lane attach failed', + expect.any(Error), + ); + expect(infoLines(h).some((l) => l.includes('connector-credential.mutated'))).toBe(false); + }); + + it('kernel:shutdown detaches lane 4, and a throwing lane-3 detach does not strand it', async () => { + const h = makeHarness({ driver: 'redis', datasourceAdmin: 'real', automation: 'real' }); + h.detachDatasource.mockImplementation(() => { throw new Error('detach exploded'); }); + await new MetadataClusterBridgePlugin().init(h.ctx); + await h.fire('kernel:ready'); + await h.fire('kernel:shutdown'); + + expect(h.detachCredential).toHaveBeenCalledTimes(1); + expect(h.logger.error).toHaveBeenCalled(); + + // Idempotent: a second shutdown does not detach twice. + await h.fire('kernel:shutdown'); + expect(h.detachCredential).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 643fab836ba..c82078dd004 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`. * - * THREE lanes, late-bound independently at `kernel:ready`, because the state - * that goes stale lives in three different owners (#13331, #13805): + * FOUR lanes, late-bound independently at `kernel:ready`, because the state + * that goes stale lives in four different owners (#13331, #13805, #27428): * * 1. **Metadata service** (`attachClusterPubSub()` — `metadata.changed`): * replays watch events into peer `MetadataManager` caches @@ -43,6 +43,16 @@ import { isInProcessClusterDriver } from './split-brain-guard.js'; * `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. + * 4. **Automation service** (`attachConnectorCredentialPubSub()` — + * `connector-credential.mutated`): fans a connector credential written or + * cleared through `PUT|DELETE /api/v1/automation/connectors/credential-refs/:ref` + * out to peers, which reconcile their declarative connector instances + * from their OWN read of the shared credential rows. The same family + * (#27428, ruled 2026-10-30 — letter A, attached here so the one place + * that enumerates cluster lanes stays one place): without it a corrected + * credential took effect on the replica that took the request, and an + * instance its upstream had refused stayed held on every other replica + * until a metadata reload or a 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 @@ -53,7 +63,9 @@ import { isInProcessClusterDriver } from './split-brain-guard.js'; * `metadata.mutated` — payload shape defined by * `ClusterMetadataMutationPayload` in `@objectstack/metadata-protocol`; * `datasource.mutated` — payload shape defined by - * `ClusterDatasourceMutationPayload` in `@objectstack/service-datasource`. + * `ClusterDatasourceMutationPayload` in `@objectstack/service-datasource`; + * `connector-credential.mutated` — payload shape defined by + * `ClusterConnectorCredentialMutationPayload` in `@objectstack/service-automation`. * * See `content/docs/kernel/cluster.mdx` §6.2. */ @@ -65,6 +77,7 @@ export class MetadataClusterBridgePlugin implements Plugin { private detach?: () => void; private detachMutation?: () => void; private detachDatasource?: () => void; + private detachConnectorCredential?: () => void; async init(ctx: PluginContext): Promise { ctx.hook('kernel:ready', async () => { @@ -80,6 +93,7 @@ export class MetadataClusterBridgePlugin implements Plugin { this.attachMetadataServiceLane(ctx, cluster); this.attachProtocolLane(ctx, cluster); this.attachDatasourceLane(ctx, cluster); + this.attachConnectorCredentialLane(ctx, cluster); }); ctx.hook('kernel:shutdown', async () => { @@ -110,6 +124,15 @@ export class MetadataClusterBridgePlugin implements Plugin { ); } this.detachDatasource = undefined; + try { + this.detachConnectorCredential?.(); + } catch (err) { + ctx.logger.error( + 'MetadataClusterBridgePlugin: connector-credential-lane detach error', + err as Error, + ); + } + this.detachConnectorCredential = undefined; }); } @@ -297,4 +320,61 @@ export class MetadataClusterBridgePlugin implements Plugin { ); } } + + /** + * Lane 4 — the automation service's `connector-credential.mutated` + * fan-out (#27428): the connector credential door adopting the family + * lanes 2 and 3 established. + * + * Duck-typed exactly like lanes 1 to 3 feature-detect their seams: this + * package must not depend on `@objectstack/service-automation`. The + * automation service publishes on the credential door it already owns and + * converges through the reconcile it already runs. + * + * Guarded on {@link isInProcessClusterDriver} from birth, like lanes 2 and + * 3: 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 attachConnectorCredentialLane(ctx: PluginContext, cluster: IClusterService): void { + let automation: unknown; + try { + automation = ctx.getService('automation'); + } catch { + ctx.logger.debug( + 'MetadataClusterBridgePlugin: no "automation" service registered, skipping connector-credential fan-out', + ); + return; + } + + const attach = (automation as { attachConnectorCredentialPubSub?: unknown }) + .attachConnectorCredentialPubSub; + if (typeof attach !== 'function') { + ctx.logger.debug( + 'MetadataClusterBridgePlugin: automation service does not expose attachConnectorCredentialPubSub(), skipping connector-credential fan-out', + ); + return; + } + + if (isInProcessClusterDriver(cluster.driver)) { + ctx.logger.debug( + `MetadataClusterBridgePlugin: cluster driver "${cluster.driver}" is in-process; connector-credential fan-out has no peers to reach, skipping`, + ); + return; + } + + try { + this.detachConnectorCredential = (attach as ( + pubsub: IClusterService['pubsub'], + nodeId: string, + ) => () => void).call(automation, cluster.pubsub, cluster.nodeId); + ctx.logger.info( + `MetadataClusterBridgePlugin: bridged connector-credential.mutated → cluster.pubsub (node=${cluster.nodeId})`, + ); + } catch (err) { + ctx.logger.error( + 'MetadataClusterBridgePlugin: connector-credential-lane attach failed', + err as Error, + ); + } + } }