Skip to content

Commit ab75996

Browse files
test(service-automation): measure a connector definition published on one replica across two replicas (#27977) (#28010)
<!-- claude-projects-attribution: {"github_login":"os-elon-musk"} --> _Requested by **Elon** · [project thread](https://claude.ai/code/project/chan_01WedDzV26nniwxv7jC88B9D?thread=cmsg_01WedDzV26nniwxv7jC88B9DU7tsm5bu7hCcGfaFJtkLYA)_ Fixes #27977 domain:server#3 · os-elon-musk · session_016WV4cx4GVsYbtK2oE4scBf Clause-②: no Metadata-protocol: none ## 维护者速读 - 改了什么:只加一个测试文件 `connector-definition-cluster.test.ts`,用两个内核(各自 ObjectQLPlugin 元数据协议 + 自动化插件)共享一个 SQLite 文件和一条总线,测量「在 A 上发布改过的连接器定义,B 会不会跟着重建」。不改任何运行时代码。 - 为什么改:#27428 的裁决 6513279565 要求单独测量这一点。结论:B 不跟。lane 2 把变更送到了 B 的协议监听器,但自动化插件的监听只认 `flow`,`metadata:reloaded` 又只在本内核触发,所以 B 一直用旧定义,直到重载或重启。缺口以 `it.fails` 钉住,修复卡是 #28002。 - 风险与回滚:零运行时风险,测试不进发布包(`files` 只发 `dist`)。回滚即删掉这个文件。 - 先看:`packages/services/service-automation/src/connector-definition-cluster.test.ts`,`it.fails` 那一例和其后的对照例。 - 两分钟验证:`pnpm --filter @objectstack/service-automation exec vitest run src/connector-definition-cluster.test.ts`,预期 `2 passed | 1 expected fail (3)`。 - 需要你做的:无,席位落地。 ## What changes One new test file, `packages/services/service-automation/src/connector-definition-cluster.test.ts`, built like `connector-credential-cluster.test.ts` (#27941): two `ObjectKernel` replicas, each running `ObjectQLPlugin`'s metadata protocol and `AutomationServicePlugin`, over ONE better-sqlite3 file both open, joined by a bus that delivers every publish to every subscriber (publisher included). Each replica is attached the way `MetadataClusterBridgePlugin` attaches it at `kernel:ready`: lane 2 (`attachMetadataMutationPubSub`) and lane 4 (`attachConnectorCredentialPubSub`). Lane 1 (`metadata.changed`) is not attached: it replays into a `MetadataManager`'s caches, and the connector reconcile reads the ObjectQL registry and the protocol's execution view, never that manager. A definition change is published on A through the per-item door (a draft `saveMetaItem`, then `publishMetaItem`), changing `providerConfig.endpoint` from V1 to V2. ## Measurement | | result | |:--|:--| | A, the publishing replica | re-materializes with V2 (ObjectQLPlugin announces A's own `metadata:reloaded` after the per-item publish) | | lane 2 to B | delivered: B's protocol replays `connector/NAME` into B's mutation listeners | | B's live instance, no reload on B | **does not converge**: B's provider is never called (`expected undefined to be 'https://v2.billing.example.test'`) | | control: `metadata:reloaded` on B | B materializes V2 from its own read of the shared store | Cause, read from the code: `subscribeFlowMutations` in `plugin.ts` returns early for any mutation whose type is not `flow`, and `metadata:reloaded` is a kernel-local hook a peer never receives. So the gap is the trigger, not the store or the read. The lane choice for the fix belongs to #28002. ## Tests - `connector-definition-cluster.test.ts`: `Tests 2 passed | 1 expected fail (3)`. - Harness health, a plain `it`: A converges and lane 2 reaches B. This keeps the `it.fails` from passing on a broken harness. - The gap, `it.fails` naming #27977 and #28002: B converges with no reload on B. Its failure was read with the case temporarily run as a plain `it`: the one assertion that fails is B's last endpoint, `undefined` instead of V2. - Control, a plain `it`: B holds no instance before the reload, and holds V2 after it. - `pnpm --filter @objectstack/service-automation test`: `Test Files 222 passed (222) · Tests 2832 passed | 1 expected fail (2833)`, lock VERDICT command-exit 0. - `pnpm --filter @objectstack/service-automation typecheck`: lock VERDICT command-exit 0. ## Acceptance notes - Side finding, same root, measured with a throwaway probe that is not committed: a direct active save (`saveMetaItem` with no mode, what `PUT /api/v1/meta/connector/:name` calls) does not re-materialize even on the writing replica. A stayed on V1 after saving V2, and its listeners heard the mutation. The seat recorded it on #28002. - No changeset (`skip-changeset`): the package publishes only `dist`. After `pnpm --filter @objectstack/service-automation build`, `grep` for the test's names in `dist` gave 0 hits, and the positive control `AutomationServicePlugin` gave 4 files. ## Verification Head `022ad12c`, base `46943657`. <details><summary>Derived gate families and lint</summary> `node scripts/pm/dispatch-gates.mjs --commands --repo objectstack-ai/objectstack` derived 50 families over the 1 changed path. 49 exited 0. `pnpm check:dual-build-cjs-loads` exited 3 (PREREQUISITE NOT MET: it needs every package's `dist` in this worktree), so it is NOT MEASURED locally and CI runs it. `check-plugin-teardown-shape --self-test` exited 3 on the shallow clone and 0 after `git fetch --unshallow`. `--ran` reconciliation: `50 derived, 49 run, 1 NOT-MEASURED, 0 UNRUN`. `pnpm lint` exited 0. </details> ## Human Minimum - [x] 1. Schemas start as Zod; `packages/spec` holds no runtime logic. (Not touched.) - [x] 2. Every name I added uses the casing for its kind. - [x] 3. No formatter ran; `pnpm lint` passes. - [x] 4. One issue, one branch off an up-to-date `main`, one pull request. - [x] 5. A change that publishes carries a changeset: nothing published, `skip-changeset`. - [x] 6. `packages/spec` touched: not touched. - [x] 7. No hand edits to `content/docs/references/`, `packages/*/CHANGELOG.md` or `content/docs/releases/`. - [x] 8. Title, body and comments are in English. - [x] The *Before Submitting* list in CONTRIBUTING.md is done. --- _Generated by [Claude Code](https://claude.ai/code/session_016WV4cx4GVsYbtK2oE4scBf)_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) https://claude.ai/code/session_016WV4cx4GVsYbtK2oE4scBf --------- Co-authored-by: Claude <noreply@anthropic.com>
1 parent 0aac0bf commit ab75996

1 file changed

Lines changed: 219 additions & 0 deletions

File tree

Lines changed: 219 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,219 @@
1+
// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
2+
3+
/**
4+
* [#27977] Cross-replica reach of a connector DEFINITION published through the
5+
* per-item publish door — measured over the topology of
6+
* `connector-credential-cluster.test.ts`: two kernels ("replicas") over ONE
7+
* shared database, each with its own materialized connectors, joined by a bus
8+
* that delivers every publish to every subscriber (publisher included).
9+
*
10+
* Each replica runs the real write path — ObjectKernel, ObjectQLPlugin's
11+
* metadata protocol, driver-sql on one better-sqlite3 file both replicas open
12+
* — and is attached the way `MetadataClusterBridgePlugin` attaches it at
13+
* `kernel:ready`: lane 2 (`attachMetadataMutationPubSub`, `metadata.mutated`)
14+
* and lane 4 (`attachConnectorCredentialPubSub`). Lane 1 (`metadata.changed`)
15+
* replays into a `MetadataManager`'s caches; the connector reconcile reads the
16+
* ObjectQL registry and the protocol's execution view, never that manager, so
17+
* attaching it could not change what is read here.
18+
*
19+
* Directions declared before running:
20+
* • A publishes a changed definition: A re-materializes it (its own
21+
* `metadata:reloaded`, announced by ObjectQLPlugin) -> GREEN
22+
* • lane 2 carries the mutation to B: B's protocol replays it into B's
23+
* mutation listeners -> GREEN
24+
* • B's live instance follows the new definition with no reload on B
25+
* -> RED
26+
* (the automation plugin's mutation listener takes `flow` only, and
27+
* `metadata:reloaded` is a kernel-local hook; pinned as `it.fails`, the
28+
* fix is #28002)
29+
* • control: a `metadata:reloaded` on B converges B from its OWN read of the
30+
* shared store, so the gap is the trigger and not the store or the read
31+
* -> GREEN
32+
*/
33+
import { mkdtempSync, rmSync } from 'node:fs';
34+
import { tmpdir } from 'node:os';
35+
import { join } from 'node:path';
36+
import { afterEach, describe, expect, it, vi } from 'vitest';
37+
import { ObjectKernel } from '@objectstack/core';
38+
import { ObjectQLPlugin, type ObjectQL } from '@objectstack/objectql';
39+
import { SqlDriver } from '@objectstack/driver-sql';
40+
import type { IPubSub, PubSubHandler } from '@objectstack/spec/contracts';
41+
import type { Connector, ConnectorProviderContext, ConnectorProviderFactory } from '@objectstack/spec/integration';
42+
43+
import { AutomationServicePlugin } from './plugin.js';
44+
import type { AutomationEngine } from './engine.js';
45+
import { installDefaultOrganization } from './default-organization.testkit.js';
46+
47+
const NAME = 'billing_27977';
48+
const V1 = 'https://v1.billing.example.test';
49+
const V2 = 'https://v2.billing.example.test';
50+
51+
const flush = () => new Promise<void>((r) => setTimeout(r, 0));
52+
53+
/** A provider-bound connector instance; `endpoint` is the definition change under test. */
54+
const definition = (endpoint: string) => ({
55+
name: NAME,
56+
label: 'Billing',
57+
type: 'api',
58+
provider: 'fake',
59+
providerConfig: { endpoint },
60+
});
61+
62+
/** The protocol surface this suite drives: the doors `PUT /meta/:type/:name` and `POST …/publish` call. */
63+
interface Protocol {
64+
saveMetaItem(request: { type: string; name: string; item: unknown; mode?: 'draft' | 'publish' }): Promise<unknown>;
65+
publishMetaItem(request: { type: string; name: string }): Promise<unknown>;
66+
onMetadataMutation(listener: (evt: { type: string; name: string; state?: string }) => void): () => void;
67+
attachMetadataMutationPubSub(pubsub: IPubSub, nodeId: string): () => void;
68+
}
69+
70+
/** One replica's upstream: records each materialization's endpoint. */
71+
function fakeProvider() {
72+
const endpoints: unknown[] = [];
73+
const factory: ConnectorProviderFactory = (ctx: ConnectorProviderContext) => {
74+
endpoints.push((ctx.providerConfig as Record<string, unknown> | undefined)?.endpoint);
75+
return {
76+
def: {
77+
name: ctx.name,
78+
label: ctx.label,
79+
type: 'api',
80+
authentication: { type: 'none' },
81+
actions: [{ key: 'request', label: 'Request' }],
82+
} as unknown as Connector,
83+
handlers: { request: async () => ({ status: 200, ok: true, body: {} }) },
84+
};
85+
};
86+
return { factory, endpoints, last: () => endpoints[endpoints.length - 1] };
87+
}
88+
89+
/** The shared bus: every publish reaches every subscriber on the channel, publisher included, one hop later. */
90+
function sharedBus(): IPubSub {
91+
const handlers = new Map<string, Set<PubSubHandler<any>>>();
92+
return {
93+
async publish(channel, payload) {
94+
for (const h of [...(handlers.get(channel) ?? [])]) {
95+
setTimeout(() => void h({ channel, payload, publishedAt: Date.now() }), 0);
96+
}
97+
},
98+
subscribe(channel, handler) {
99+
const set = handlers.get(channel) ?? new Set();
100+
set.add(handler as PubSubHandler<any>);
101+
handlers.set(channel, set);
102+
return () => set.delete(handler as PubSubHandler<any>);
103+
},
104+
async close() {},
105+
};
106+
}
107+
108+
const kernels: ObjectKernel[] = [];
109+
const dirs: string[] = [];
110+
afterEach(async () => {
111+
for (const k of kernels.splice(0)) {
112+
if (k.getState() === 'running') await k.shutdown().catch(() => {});
113+
}
114+
for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true });
115+
vi.restoreAllMocks();
116+
});
117+
118+
/** The ONE database file both replicas open. */
119+
function sharedDatabaseFile(): string {
120+
const dir = mkdtempSync(join(tmpdir(), 'os-27977-'));
121+
dirs.push(dir);
122+
return join(dir, 'shared.sqlite');
123+
}
124+
125+
/** One replica over the shared file, attached to `bus` the way the cluster bridge attaches it. */
126+
async function replica(file: string, nodeId: string, bus: IPubSub) {
127+
const provider = fakeProvider();
128+
let hooks: { trigger(event: string, payload?: unknown): Promise<void> } | undefined;
129+
const kernel = new ObjectKernel({ logger: { level: 'fatal' }, gracefulShutdown: false } as never);
130+
kernels.push(kernel);
131+
await kernel.use(new ObjectQLPlugin());
132+
await kernel.use(new AutomationServicePlugin({ suspendedRunStore: 'memory', credentialResolver: () => undefined }));
133+
await kernel.use({
134+
name: `test.harness.${nodeId}`,
135+
type: 'standard' as const,
136+
version: '1.0.0',
137+
dependencies: ['com.objectstack.service-automation'],
138+
async init(ctx: any) {
139+
hooks = ctx;
140+
ctx.getService('automation').registerConnectorProvider('fake', provider.factory);
141+
},
142+
async start() {},
143+
} as never);
144+
await kernel.bootstrap();
145+
146+
const ql = kernel.getService<ObjectQL>('objectql');
147+
const driver = new SqlDriver({ client: 'better-sqlite3', connection: { filename: file }, useNullAsDefault: true });
148+
await driver.connect();
149+
ql.registerDriver(driver as never, true);
150+
await ql.syncSchemas();
151+
await installDefaultOrganization(ql);
152+
153+
const protocol = kernel.getService<Protocol>('protocol');
154+
const automation = kernel.getService<AutomationEngine>('automation');
155+
// What `MetadataClusterBridgePlugin`'s lanes 2 and 4 do at `kernel:ready` on a cross-process driver.
156+
protocol.attachMetadataMutationPubSub(bus, nodeId);
157+
automation.attachConnectorCredentialPubSub(bus, nodeId);
158+
159+
const heard: string[] = [];
160+
protocol.onMetadataMutation((evt) => heard.push(`${evt.type}/${evt.name}`));
161+
162+
/** The per-item publish door: a draft save, then `POST /api/v1/meta/connector/:name/publish`. */
163+
const publish = async (item: unknown) => {
164+
await protocol.saveMetaItem({ type: 'connector', name: NAME, item, mode: 'draft' });
165+
await protocol.publishMetaItem({ type: 'connector', name: NAME });
166+
};
167+
const reload = () => hooks!.trigger('metadata:reloaded', { changed: [`connector/${NAME}`] });
168+
const live = () => automation.getRegisteredConnectors().includes(NAME);
169+
return { provider, publish, reload, live, heard };
170+
}
171+
172+
describe('[#27977] a connector definition published on one replica', () => {
173+
it('re-materializes on the publishing replica, and lane 2 carries the mutation to the peer', async () => {
174+
const file = sharedDatabaseFile();
175+
const bus = sharedBus();
176+
const a = await replica(file, 'node-a', bus);
177+
const b = await replica(file, 'node-b', bus);
178+
179+
await a.publish(definition(V1));
180+
expect(a.live()).toBe(true);
181+
expect(a.provider.last()).toBe(V1);
182+
183+
await a.publish(definition(V2));
184+
expect(a.provider.last()).toBe(V2);
185+
await vi.waitFor(() => expect(b.heard).toContain(`connector/${NAME}`));
186+
});
187+
188+
// Measured by #27977; the fix is #28002. Promote to a plain `it` when #28002 lands.
189+
it.fails('[#27977] converges the peer with no reload on the peer — known gap, filed as #28002', async () => {
190+
const file = sharedDatabaseFile();
191+
const bus = sharedBus();
192+
const a = await replica(file, 'node-a', bus);
193+
const b = await replica(file, 'node-b', bus);
194+
195+
await a.publish(definition(V1));
196+
await a.publish(definition(V2));
197+
await vi.waitFor(() => expect(b.heard).toContain(`connector/${NAME}`));
198+
await vi.waitFor(() => expect(b.provider.last()).toBe(V2), { timeout: 500 });
199+
});
200+
201+
it('control: a reload on the peer converges it from its own read of the shared store', async () => {
202+
const file = sharedDatabaseFile();
203+
const bus = sharedBus();
204+
const a = await replica(file, 'node-a', bus);
205+
const b = await replica(file, 'node-b', bus);
206+
207+
await a.publish(definition(V1));
208+
await a.publish(definition(V2));
209+
await vi.waitFor(() => expect(b.heard).toContain(`connector/${NAME}`));
210+
for (let i = 0; i < 5; i++) await flush();
211+
// The measured state before the reload: the peer never materialized the instance.
212+
expect(b.live()).toBe(false);
213+
expect(b.provider.endpoints).toEqual([]);
214+
215+
await b.reload();
216+
expect(b.live()).toBe(true);
217+
expect(b.provider.last()).toBe(V2);
218+
});
219+
});

0 commit comments

Comments
 (0)