fix(runtime): a no-organization job pull runs as the organization its writes are stamped with

.changeset/31823-job-pull-organization.md

@@ -0,0 +1,13 @@
+---
+'@objectstack/runtime': patch
+'@objectstack/objectql': patch
+---
+
+fix(runtime): a job with no `organization` pulls as the organization its writes are stamped with, so its upsert match uses the organization index
+
+Clause-②: yes (widening) — @objectstack/objectql gains `ObjectQL#systemWriteOrganization()`, the engine's own answer to which organization an organization-less system write resolves to
+Metadata-protocol: none
+
+A `job` whose `pull` syncs a mapping and that declares no `organization` ran its pull with no organization in its context (`{ isSystem: true }`). Every write of that pull was still stamped with the install's organization: on the `single` posture with exactly one organization, the engine derives it for an organization-less system write (ADR-0131 D9). But the import's match lookup had no organization to scope by, so on `single`, where a read gets no tenant scope, it read the target table once per row instead of probing the `(organization_id, <key>)` unique index. A large sync on the most common setup stayed table-scan slow.
+
+Each run of such a pull now asks the engine, once, before it reads or writes, the same decision the engine's system insert makes, and runs as that organization when the engine derives one. The lookup then carries it on the tenant column, and the rows land exactly where they did before. When the engine does not derive one (a `group` or `isolated` posture, zero or several organizations, a composition with no organization object), the pull runs as before, and its writes are refused or stamped exactly as before. A job that declares an `organization` is unchanged, and so are the `body` and `handler` run forms. If the organization probe fails, the run says so once at `warn`, naming the job, and goes on as before.

packages/objectql/src/engine-middleware-and-context.ts

@@ -40,7 +40,7 @@ import type { ExecutionContext } from '@objectstack/spec/kernel';
 import { isWriteEpochOperation } from './write-epoch.js';
 import { bridgeAuthzInvalidation } from './authz-invalidation-bridge.js';
 import { isMissingTableError } from '@objectstack/metadata/errors';
-import { ORGANIZATION_OBJECT, SystemWriteOrganizationRequiredError, carriesOrganization, resolveSystemWriteOrganization, resolveTenantFieldName } from './tenancy/system-write-organization.js';
+import { ORGANIZATION_OBJECT, SystemWriteOrganizationRequiredError, carriesOrganization, resolveSystemWriteOrganization, resolveTenantFieldName, type SystemWriteOrganizationDecision } from './tenancy/system-write-organization.js';
 import { isFederatedObject } from './federated-object.js';
 import type { EngineMiddleware, OperationContext } from './engine-top-level.js';
 import type { ObjectQL } from './engine.js';
@@ -359,24 +359,34 @@ export class MiddlewareAndContextFacet {
     if (rows.every((row) => carriesOrganization(row?.[tenantField]))) return undefined;
 
     const posture = this.resolveEnginePosture();
+    const decision = await MiddlewareAndContextFacet.systemWriteOrganization.call(this, posture);
+    if (decision.kind === 'derived') return decision.organizationId;
+    if (decision.kind === 'no-organization-object') return undefined;
+    throw new SystemWriteOrganizationRequiredError(
+      object,
+      posture,
+      decision.reason,
+      decision.organizationCount,
+    );
+  }
+
+  /**
+   * The body of `ObjectQL#systemWriteOrganization`; the method's docblock there
+   * is its contract. Also the decision half of
+   * {@link resolveSystemInsertOrganization}, which hands in the posture it has
+   * already read, so the insert and the public answer are one assembly.
+   */
+  static async systemWriteOrganization(this: Pick<MiddlewareAndContextHost, '_registry' | 'probeInstallOrganizations' | 'resolveEnginePosture'>, posture = this.resolveEnginePosture()): Promise<SystemWriteOrganizationDecision> {
     // [ADR-0131 D9] Zero organizations is refused under `single` only where
     // the composition registers the organization object; the composition that
     // registers none keeps today's branch (`no-organization-object`). The
     // probe itself is unchanged and answers `[]` for both — this is the one
     // fact that tells them apart, read from the same registry the probe reads.
-    const decision = await resolveSystemWriteOrganization({
+    return resolveSystemWriteOrganization({
       posture,
       probeOrganizations: () => this.probeInstallOrganizations(),
       organizationObjectRegistered: !!this._registry.getObject(ORGANIZATION_OBJECT),
     });
-    if (decision.kind === 'derived') return decision.organizationId;
-    if (decision.kind === 'no-organization-object') return undefined;
-    throw new SystemWriteOrganizationRequiredError(
-      object,
-      posture,
-      decision.reason,
-      decision.organizationCount,
-    );
   }
 
   /**

packages/objectql/src/engine.ts

@@ -200,6 +200,7 @@ import { redactBoundStatement } from '@objectstack/types';
 // supplies the facts and turns a `refuse` into the throw.
 import {
   ORGANIZATION_OBJECT,
+  type SystemWriteOrganizationDecision,
 } from './tenancy/system-write-organization.js';
 import {
   type TenancyPosture,
@@ -3054,6 +3055,32 @@ export class ObjectQL implements IObjectQLEngine {
     return SecretsAndRoutingFacet.setTenancyPostureProvider.call(secretsAndRoutingHost.of(this), provider);
   }
 
+  /**
+   * [ADR-0131 D9] The organization a SYSTEM-context write that carries none
+   * resolves to on this engine — the decision its own system insert makes
+   * (`resolveSystemInsertOrganization`), from the same three inputs: the
+   * posture in force (the injected tenancy-posture accessor, else the
+   * deployment's), the install's organizations (the memoised probe, capped at
+   * two), and whether this composition registers `sys_organization` at all.
+   * The predicate is `@objectstack/core`'s `resolveSystemWriteOrganization`,
+   * never a second spelling of it.
+   *
+   * For a caller that runs system work with no organization of its own and
+   * wants to carry, ahead of its writes, the organization those writes would
+   * be stamped with — a job that declares no `organization` resolves its pull's
+   * context here, so the pull's reads scope by the organization its writes
+   * land in. Not a write and not a gate: it refuses nothing itself. A `refuse`
+   * decision is the answer the write would turn into its refusal, and the
+   * write stays the one that refuses.
+   *
+   * A probe that could not run (an outage, not an empty table) propagates, as
+   * it does on the insert.
+   */
+  systemWriteOrganization(): Promise<SystemWriteOrganizationDecision> {
+    // [#31823] The body lives in `engine-middleware-and-context.ts` (see `MiddlewareAndContextHost`).
+    return MiddlewareAndContextFacet.systemWriteOrganization.call(middlewareAndContextHost.of(this));
+  }
+
   private async encryptSecretFields(
     object: string,
     row: Record<string, unknown>,

packages/objectql/src/system-write-organization.test.ts

@@ -229,6 +229,55 @@ describe('#8844 the decision — the ruling as one pure function', () => {
   });
 });
 
+describe('ObjectQL#systemWriteOrganization — the engine answers the decision its system insert makes', () => {
+  // A caller with no organization of its own (a job pull) asks this up front,
+  // so the organization its reads scope by is the one its writes are stamped
+  // with. It must be the insert's own assembly: same posture, same probe (and
+  // memo), same registry fact — and it writes nothing.
+  const probes = (observed: ObservedCall[]) => observed.filter((c) => c.object === 'sys_organization' && c.method === 'find').length;
+
+  it('single, one organization: derived — the organization a system insert is then stamped with', async () => {
+    const { engine, observed } = await makeEngine({ posture: 'single' });
+    expect(await engine.systemWriteOrganization()).toEqual({ kind: 'derived', organizationId: ORG_ID });
+    expect(observed.filter((c) => c.method !== 'find')).toEqual([]);
+
+    await engine.insert('dispatch_order', { subject: 'stamped' }, { context: SYSTEM_CTX } as any);
+    expect(lastWrite(observed, 'dispatch_order')?.options?.tenantId).toBe(ORG_ID);
+    // One probe for both: the answer and the insert share the engine's memo.
+    expect(probes(observed)).toBe(1);
+  });
+
+  it('single, several organizations: refuse (ambiguous), as the insert refuses', async () => {
+    const { engine } = await makeEngine({ posture: 'single', organizations: [ORG_ID, SECOND_ORG_ID] });
+    expect(await engine.systemWriteOrganization())
+      .toEqual({ kind: 'refuse', reason: 'ambiguous-organization', organizationCount: 2 });
+  });
+
+  it.each(['group', 'isolated'] as const)('the injected %s posture: refuse (walled), and nothing is probed', async (posture) => {
+    const { engine, observed } = await makeEngine({ posture });
+    expect(await engine.systemWriteOrganization()).toEqual({ kind: 'refuse', reason: 'walled-posture' });
+    expect(probes(observed)).toBe(0);
+  });
+
+  it('no provider injected: the posture is the deployment\'s, as on the insert', async () => {
+    vi.stubEnv('OS_TENANCY_POSTURE', 'isolated');
+    const { engine } = await makeEngine();
+    expect(await engine.systemWriteOrganization()).toEqual({ kind: 'refuse', reason: 'walled-posture' });
+  });
+
+  it('a composition with no organization object: no-organization-object', async () => {
+    const { engine } = await makeEngine({ posture: 'single', lean: true });
+    expect(await engine.systemWriteOrganization()).toEqual({ kind: 'no-organization-object' });
+  });
+
+  it('a probe that cannot run propagates — it is not answered as "no organization"', async () => {
+    const { engine } = await makeEngine({ posture: 'single' });
+    const driver = engine.getDriverForObject('dispatch_order') as any;
+    driver.find = async () => { throw new Error('connection reset'); };
+    await expect(engine.systemWriteOrganization()).rejects.toThrow('connection reset');
+  });
+});
+
 describe('#8844 single-tenant — the system write is stamped like a session write', () => {
   it('derives the install organization for a system-context insert, and a session write is unchanged', async () => {
     const { engine, observed } = await makeEngine();