fix(service-datasource): one live introspection per datasource per validation sweep

.changeset/validate-sweep-introspection-memo.md

@@ -0,0 +1,6 @@
+---
+'@objectstack/service-datasource': patch
+'@objectstack/spec': patch
+---
+
+`validateAll`/`validateDatasource` now read each datasource's live schema once per sweep instead of once per federated object: the sweep threads a per-call introspection memo through the validation body, so M objects on one datasource cost one remote introspection round-trip (a rejected read is shared the same way — one connection attempt, M failure rows). The memo lives and dies inside a single call, so a long-lived service never serves a stale schema to a later sweep, and direct `validateObject` calls still read live every time. The `IExternalDatasourceService.validateAll` docstring, which promised "parallelised per datasource" while the implementation parallelised per object, now states the actual behaviour.

packages/services/service-datasource/src/__tests__/external-datasource-service.test.ts

@@ -672,6 +672,157 @@ describe('validateDatasource', () => {
  });
 });
 
+/**
+ * [#10962] Per-sweep introspection memo — the CALL COUNT is the deliverable.
+ *
+ * `validateObject` reads the live remote schema on every call, so a sweep over
+ * M federated objects on one datasource used to perform M concurrent
+ * `introspect(datasource)` round-trips against the same remote. The fix
+ * threads a per-call memo through `validateAll`/`validateDatasource`: one live
+ * read per datasource per sweep.
+ *
+ * The memo's LIFETIME is the other half of the contract, and the harder one:
+ * per call, never per instance. A `Map` cached on the service would pass every
+ * counting assertion a per-call memo passes — only the second-sweep cases
+ * below (a fresh read per sweep, and a remote change visible to the next
+ * sweep) tell them apart. Do not weaken those to "at least once".
+ */
+describe('per-sweep introspection memo [#10962]', () => {
+ const M_DATASOURCE = 'wh';
+ const SIDE_DATASOURCE = 'wh_b';
+
+ /** M > 1 objects on one datasource — the population the memo collapses. */
+ const OBJECTS: ObjectLike[] = [
+ ...['wh_orders_a', 'wh_orders_b', 'wh_orders_c'].map((name) => ({
+ name,
+ datasource: M_DATASOURCE,
+ external: { remoteName: 'orders' },
+ fields: { order_id: { type: 'text' } },
+ })),
+ {
+ name: 'side_orders',
+ datasource: SIDE_DATASOURCE,
+ external: { remoteName: 'orders' },
+ fields: { order_id: { type: 'text' } },
+ },
+ ];
+
+ function makeCounting(opts: { unreachable?: readonly string[] } = {}) {
+ const introspected: string[] = [];
+ const unreachable = new Set(opts.unreachable ?? []);
+ let tables: IntrospectedSchema['tables'] = {
+ orders: {
+ name: 'orders',
+ indexes: [],
+ columns: [{ name: 'order_id', type: 'text', nullable: false, primaryKey: true }],
+ },
+ };
+ const svc = new ExternalDatasourceService({
+ introspect: async (datasource: string) => {
+ introspected.push(datasource);
+ if (unreachable.has(datasource)) throw new Error(`connect ECONNREFUSED (${datasource})`);
+ return { dialect: 'postgres', introspectedAt: '2026-08-23T00:00:00.000Z', tables };
+ },
+ getDatasource: async (name: string) =>
+ [M_DATASOURCE, SIDE_DATASOURCE].includes(name) ? { name, schemaMode: 'external' } : undefined,
+ getObject: async (name: string) => OBJECTS.find((o) => o.name === name),
+ listObjects: async () => OBJECTS,
+ logger: { warn: () => {} },
+ });
+ return {
+ svc,
+ introspected,
+ /** Simulate the remote dropping its tables between sweeps. */
+ dropRemoteTables: () => {
+ tables = {};
+ },
+ };
+ }
+
+ it('validateAll reads each datasource once per sweep — M objects, one live read', async () => {
+ // The claim is only non-vacuous when M really exceeds 1.
+ expect(OBJECTS.filter((o) => o.datasource === M_DATASOURCE).length).toBeGreaterThan(1);
+ const { svc, introspected } = makeCounting();
+
+ const report = await svc.validateAll();
+
+ expect([...introspected].sort()).toEqual([M_DATASOURCE, SIDE_DATASOURCE]);
+ expect(report.ok).toBe(true);
+ expect(report.results).toHaveLength(OBJECTS.length);
+ });
+
+ it('validateDatasource reads its datasource once for M objects', async () => {
+ const { svc, introspected } = makeCounting();
+
+ const report = await svc.validateDatasource(M_DATASOURCE);
+
+ expect(introspected).toEqual([M_DATASOURCE]);
+ expect(report.ok).toBe(true);
+ expect(report.results).toHaveLength(3);
+ });
+
+ it('a second sweep reads live again — the memo is per call, not a service-instance cache', async () => {
+ const { svc, introspected, dropRemoteTables } = makeCounting();
+
+ const first = await svc.validateAll();
+ expect(first.ok).toBe(true);
+ expect(introspected.filter((d) => d === M_DATASOURCE)).toHaveLength(1);
+
+ // The remote changes between sweeps. A per-instance cache would keep the
+ // counting pin above green while answering this sweep from last sweep's
+ // schema — stale `ok: true` — which is exactly what this case refuses.
+ dropRemoteTables();
+ const second = await svc.validateAll();
+
+ expect(introspected.filter((d) => d === M_DATASOURCE)).toHaveLength(2);
+ expect(second.ok).toBe(false);
+ for (const r of second.results) {
+ expect(r.ok).toBe(false);
+ expect(r.diffs[0]).toMatchObject({ kind: 'missing_table', severity: 'error' });
+ }
+ });
+
+ it('direct validateObject stays live: two calls are two reads, and a remote change is seen', async () => {
+ const { svc, introspected, dropRemoteTables } = makeCounting();
+
+ const before = await svc.validateObject('wh_orders_a');
+ expect(before.ok).toBe(true);
+
+ dropRemoteTables();
+ const after = await svc.validateObject('wh_orders_a');
+
+ expect(introspected).toEqual([M_DATASOURCE, M_DATASOURCE]);
+ expect(after.ok).toBe(false);
+ expect(after.diffs[0]).toMatchObject({ kind: 'missing_table', severity: 'error' });
+ });
+
+ it('one unreachable remote costs ONE connection attempt and still yields a failure row per object', async () => {
+ const { svc, introspected } = makeCounting({ unreachable: [M_DATASOURCE] });
+
+ const report = await svc.validateDatasource(M_DATASOURCE);
+
+ expect(introspected).toEqual([M_DATASOURCE]);
+ expect(report.ok).toBe(false);
+ expect(report.results).toHaveLength(3);
+ for (const r of report.results) {
+ expect(r).toMatchObject({
+ ok: false,
+ datasource: M_DATASOURCE,
+ diffs: [
+ expect.objectContaining({
+ // [#11166] A throw out of introspect is `unreachable`, never an
+ // invented `missing_table` — this pin is about the COUNT (one
+ // shared connection attempt), and rides the kind ruling as-is.
+ kind: 'unreachable',
+ severity: 'error',
+ actual: `connect ECONNREFUSED (${M_DATASOURCE})`,
+ }),
+ ],
+ });
+ }
+ });
+});
+
 describe('refreshCatalog', () => {
  it('produces a snapshot with suggested field types', async () => {
  const svc = makeService();

packages/services/service-datasource/src/external-datasource-service.ts

@@ -593,6 +593,25 @@ export class ExternalDatasourceService implements IExternalDatasourceService {
  }
 
  async validateObject(objectName: string): Promise<SchemaValidationResult> {
+ // A direct call performs its own live read — no memo. Read reuse is the
+ // sweep's per-call concern ({@link validateEach}), never this method's: a
+ // long-lived service must answer every direct call from the remote's
+ // schema as it is NOW (pinned: two direct calls are two live reads).
+ return this.validateObjectUsing(objectName, (ds) => this.config.introspect(ds));
+ }
+
+ /**
+ * [#10962] The body of {@link validateObject}, with the live-schema read
+ * abstracted behind `readSchema` so one sweep can share a single read per
+ * datasource across all of its objects. `readSchema` is either
+ * `config.introspect` itself (the public single-object path above) or the
+ * per-sweep memoised reader from {@link sweepScopedIntrospect} — never a
+ * cache that outlives one call.
+ */
+ private async validateObjectUsing(
+ objectName: string,
+ readSchema: (datasource: string) => Promise<IntrospectedSchema>,
+ ): Promise<SchemaValidationResult> {
  const obj = await this.config.getObject(objectName);
  if (!obj) {
  throw new Error(`Object '${objectName}' not found.`);
@@ -605,7 +624,7 @@ export class ExternalDatasourceService implements IExternalDatasourceService {
  return { ok: true, datasource, object: objectName, diffs: [] };
  }
 
- const schema = await this.config.introspect(datasource);
+ const schema = await readSchema(datasource);
  const dialect = schema.dialect as SqlDialect | undefined;
  const remoteName = obj.external?.remoteName ?? obj.name;
  const table = this.findTable(schema, remoteName);
@@ -684,6 +703,34 @@ export class ExternalDatasourceService implements IExternalDatasourceService {
  return o.external !== undefined || Boolean(o.datasource && o.datasource !== 'default');
  }
 
+ /**
+ * [#10962] One live schema read per datasource per SWEEP.
+ *
+ * Returns a reader that memoises `config.introspect` by datasource name for
+ * the lifetime of ONE {@link validateEach} call. The memo is a local of that
+ * call — deliberately NOT an instance field — so a long-lived service can
+ * never serve a stale schema to a later sweep: the next `validateAll()` /
+ * `validateDatasource()` builds a fresh memo and reads live again (both
+ * directions pinned in `__tests__/external-datasource-service.test.ts`).
+ *
+ * The PROMISE is memoised, not the resolved value: the sweep validates its
+ * objects concurrently (`Promise.all`), so the first reader for a datasource
+ * starts the read and every concurrent sibling awaits the same in-flight
+ * promise. A rejected read is shared the same way — M objects on one
+ * unreachable datasource produce M failure rows from ONE connection attempt.
+ */
+ private sweepScopedIntrospect(): (datasource: string) => Promise<IntrospectedSchema> {
+ const memo = new Map<string, Promise<IntrospectedSchema>>();
+ return (datasource) => {
+ let read = memo.get(datasource);
+ if (!read) {
+ read = this.config.introspect(datasource);
+ memo.set(datasource, read);
+ }
+ return read;
+ };
+ }
+
  /**
  * Validate a chosen set of objects, one report.
  *
@@ -692,6 +739,9 @@ export class ExternalDatasourceService implements IExternalDatasourceService {
  * one object whose definition vanished mid-sweep) must not erase the verdicts
  * of the objects that did validate.
  *
+ * [#10962] All objects in one call share one live schema read per datasource
+ * (see {@link sweepScopedIntrospect}); the memo dies with this call.
+ *
  * ## Why the row's kind is `unreachable` for EVERY throw — no error sniffing
  *
  * This catch used to invent `kind: 'missing_table'`, so a refused
@@ -719,9 +769,10 @@ export class ExternalDatasourceService implements IExternalDatasourceService {
  * mislabel in a second direction.
  */
  private async validateEach(objects: ObjectLike[]): Promise<SchemaValidationReport> {
+ const readSchema = this.sweepScopedIntrospect();
  const results = await Promise.all(
  objects.map((o) =>
- this.validateObject(o.name).catch((err): SchemaValidationResult => {
+ this.validateObjectUsing(o.name, readSchema).catch((err): SchemaValidationResult => {
  this.logger?.warn(`validateObject('${o.name}') failed`, err);
  return {
  ok: false,

packages/spec/src/contracts/external-datasource-service.ts

@@ -154,6 +154,6 @@ export interface IExternalDatasourceService {
  /** Validate one federated object against the live remote table. */
  validateObject(objectName: string): Promise<SchemaValidationResult>;
 
- /** Validate every federated object, parallelised per datasource. */
+ /** Validate every federated object in parallel; each datasource's live schema is read once per call. */
  validateAll(): Promise<SchemaValidationReport>;
 }