diff --git a/.changeset/29604-approver-directory-refusal-queue-paging.md b/.changeset/29604-approver-directory-refusal-queue-paging.md new file mode 100644 index 00000000000..f9b0dac2830 --- /dev/null +++ b/.changeset/29604-approver-directory-refusal-queue-paging.md @@ -0,0 +1,20 @@ +--- +'@objectstack/plugin-approvals': patch +'@objectstack/plugin-sharing': patch +--- + +fix(plugin-approvals)!: an approval node refuses to open when a team or position approver's directory read fails, or a queue approver's expansion rejects, and the ownership-queue reads page to completion + +Clause-②: yes (narrowing) — an approval node no longer opens when a `team` or `position` approver (static, or an `expression` approver with `resolveAs` `team` / `position`) cannot be expanded because a directory read or the cross-organization read filter failed, or when a `queue` approver's expansion rejects. Under every `onEmptyApprovers` policy, the `fallbackApprovers` slate included, it now fails the node with a retryable `APPROVER_DIRECTORY_UNREADABLE:` error naming the read, where it used to open on the rest of the slate, on the fallback slate or on the administrators, or auto-approve. The registered ownership-queue service does not reject yet: a failed member read inside it still answers nobody until #29604 part 2 (#30251), so a `queue` approver refuses here only when `expandMembers` itself rejects. Every other input answers as before. +Metadata-protocol: none + + + +**BREAKING** accept-set narrowing at the approval node (`@objectstack/plugin-approvals`): a failed directory read behind a `team` or `position` approver, or a `queue` approver whose expansion rejects, refuses the node instead of opening it on a slate that is missing people (ADR-0142 D6 as amended, ruling record 6599731698 on #29604). + +- **Before.** A `team` or `position` approver whose `sys_team_member`, `sys_user_position` or `sys_member` read failed kept its `type:value` literal slot, which nobody can act on, and logged the failure. A `queue` approver whose expansion threw answered "expanded to nobody" and added no slot. Either way the node went on as if those people did not exist: beside other approvers it opened on them alone, and on an otherwise empty slate the node's `onEmptyApprovers` policy acted, opening on the `fallbackApprovers` or the organization's administrators, or auto-approving. +- **Now.** The node does not open and writes no request, approver or action row. It fails with an `ApproverDirectoryUnreadableError` whose message starts `APPROVER_DIRECTORY_UNREADABLE:`, names the approver, the organization and the read that failed (a failure of the cross-organization read filter is named as that filter's read, on a static approver and on an `expression` one alike), and is retryable: the flow node's failure is a runtime one, so the run can be resumed once the directory can be read. A `fallbackApprovers` slate whose own `team` or `position` read fails refuses the same way. `department` / `business_unit`, `org_membership_level` and `manager` approvers are not named by the ruling and answer exactly as before. +- **Unchanged.** A decision on a request opened before the approver snapshot existed still re-reads its slate on the old path, which keeps the literal slot on a failed read, because the decision has already been recorded when that read happens. +- **Not yet.** The ownership-queue service shipped in `@objectstack/plugin-sharing` answers a failed `sys_queue` / `sys_queue_member`, team or position read with nobody (logged once) rather than rejecting, so a `queue` approver backed by it still adds no slot on such a failure. Its refusal arrives with #29604 part 2, once the contract can carry it (#30251). + +`@objectstack/plugin-sharing`: the ownership-queue service reads its `sys_queue` and `sys_queue_member` rows by id, page after page, to the end, where it read one page of 10,000 rows and answered on it. A queue's 10,001st member, an organization's 10,001st queue and the queues naming a recipient past that bound are now found. A read that fails partway answers as a failed read did before (that read contributes nobody, logged once, with the read's reason and without the walk's own "refuses instead" clause), never with the pages it had read. diff --git a/packages/plugins/plugin-approvals/src/approval-service.test.ts b/packages/plugins/plugin-approvals/src/approval-service.test.ts index 4862830550c..b656a179d72 100644 --- a/packages/plugins/plugin-approvals/src/approval-service.test.ts +++ b/packages/plugins/plugin-approvals/src/approval-service.test.ts @@ -11,11 +11,21 @@ import { describe, it, expect, beforeEach, vi } from 'vitest'; import { APPROVAL_REVISE_NODE_TYPE, templateSlotSource } from '@objectstack/spec/automation'; -import { ApprovalService, REMIND_COOLDOWN_MS, ESCALATION_ENABLED_FLIP_CUTOFF_MS, SLA_ACTOR_ID } from './approval-service.js'; +import { + ApprovalService, ApproverDirectoryUnreadableError, REMIND_COOLDOWN_MS, ESCALATION_ENABLED_FLIP_CUTOFF_MS, SLA_ACTOR_ID, +} from './approval-service.js'; import { bindApprovalLockHook, bindDelegationWriteGuard, unbindAllHooks } from './lifecycle-hooks.js'; +import { filterApproversWhoCanRead } from './approver-org-scope.js'; import { celEngine } from '@objectstack/formula'; import { SysApprovalRequest } from './sys-approval-request.object.js'; +// [#29604] A pass-through spy on the ADR-0105 D9 read filter, so one describe +// below can make it reject. Every other case runs the real filter through it. +vi.mock('./approver-org-scope.js', async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, filterApproversWhoCanRead: vi.fn(actual.filterApproversWhoCanRead) }; +}); + interface FakeRow { [k: string]: any } /** @@ -4207,6 +4217,205 @@ describe('ApprovalService — a queue approver routes to the queue\'s members (A }); }); +// [ADR-0142 D6 as amended; ruling record 6599731698 on #29604, letter B] A +// directory read behind a `team`, `position` or `queue` approver that FAILS is +// not an empty slate: the node refuses to open under every `onEmptyApprovers` +// policy, on its own retryable error class, naming the read — and writes +// nothing. Before, the slot fell back to its literal and the policy acted on a +// slate missing the people who could not be read. +describe('ApprovalService — a team, position or queue approver whose directory read fails refuses the node (ADR-0142 D6)', () => { + const POLICIES = ['admin_rescue', 'auto_approve', 'fail', 'fallback'] as const; + + /** A service over `engine` whose reads of `failing` throw, as a driver outage does. */ + const outage = (failing: string[], queues?: () => any) => { + const engine = makeFakeEngine(); + engine._tables['sys_team_member'] = [{ id: 'tm1', team_id: 't_ops', user_id: 'u5' }]; + engine._tables['sys_user_position'] = [{ id: 'up1', position: 'cfo', user_id: 'u9', organization_id: 't1' }]; + const real = engine.find.bind(engine); + engine.find = async (object: string, options?: any) => { + if (failing.includes(object)) throw new Error(`SQLITE_IOERR: disk I/O error reading ${object}`); + return real(object, options); + }; + const warnings: any[] = []; + let n = 0; + const svc = new ApprovalService({ + engine, + clock: { now: () => new Date(1757000000000 + (n++) * 1000) }, + logger: { warn: (msg: any, meta: any) => warnings.push([msg, meta]) }, + ...(queues ? { ownershipQueues: queues } : {}), + }); + return { engine, svc, warnings }; + }; + const node = (approvers: any[], configExtra: Record = {}) => ({ + ...openInput([]), + config: { approvers, behavior: 'first_response' as const, lockRecord: false, ...configExtra }, + }); + const policy = (p: (typeof POLICIES)[number]) => ( + p === 'fallback' ? { onEmptyApprovers: p, fallbackApprovers: [{ type: 'user', value: 'u_backstop' }] } : { onEmptyApprovers: p } + ); + const refusalOf = (p: Promise) => p.then(() => null, (e: unknown) => e); + const nothingWritten = (engine: ReturnType) => { + expect(engine._tables['sys_approval_request'] ?? []).toEqual([]); + expect(engine._tables['sys_approval_approver'] ?? []).toEqual([]); + expect(engine._tables['sys_approval_action'] ?? []).toEqual([]); + }; + + for (const [type, value, table] of [['team', 't_ops', 'sys_team_member'], ['position', 'cfo', 'sys_user_position']] as const) { + for (const p of POLICIES) { + it(`a ${type} approver whose ${table} read fails: onEmptyApprovers '${p}' opens nothing and approves nothing`, async () => { + const { engine, svc } = outage([table]); + const refusal = await refusalOf(svc.openNodeRequest(node([{ type, value }], policy(p)), CTX)); + + expect(refusal).toBeInstanceOf(ApproverDirectoryUnreadableError); + expect(refusal).toMatchObject({ approverType: type, approverValue: value, retryable: true }); + expect((refusal as Error).message).toMatch(new RegExp(`^APPROVER_DIRECTORY_UNREADABLE: approver '${type}:${value}'`)); + expect((refusal as ApproverDirectoryUnreadableError).failedRead).toMatch(/SQLITE_IOERR/); + expect((refusal as Error).message).not.toMatch(/NO_APPROVERS|expanded to nobody/); + nothingWritten(engine); + }); + } + } + + it('beside a staffed person, the node still refuses: no slate shortened to the people who could be read', async () => { + const { engine, svc } = outage(['sys_team_member']); + const refusal = await refusalOf(svc.openNodeRequest( + node([{ type: 'user', value: 'u2' }, { type: 'team', value: 't_ops' }]), CTX, + )); + + expect(refusal).toBeInstanceOf(ApproverDirectoryUnreadableError); + nothingWritten(engine); + }); + + it('a fallback slate whose directory read fails refuses too, rather than degrade to the administrators', async () => { + // The primary slate reads fine and is empty (an unstaffed position); the + // declared fallback names a team whose read fails. + const { engine, svc } = outage(['sys_team_member']); + const refusal = await refusalOf(svc.openNodeRequest(node([{ type: 'position', value: 'nobody_holds_this' }], { + onEmptyApprovers: 'fallback', fallbackApprovers: [{ type: 'team', value: 't_ops' }], + }), CTX)); + + expect(refusal).toBeInstanceOf(ApproverDirectoryUnreadableError); + expect(refusal).toMatchObject({ approverType: 'team', approverValue: 't_ops' }); + nothingWritten(engine); + }); + + it('a queue whose expansion fails refuses before its "expanded to nobody" answer', async () => { + const { engine, svc, warnings } = outage([], () => ({ + expandMembers: async () => { throw new Error('ownership-queue read failed'); }, + })); + const refusal = await refusalOf(svc.openNodeRequest(node([{ type: 'queue', value: 'billing' }]), CTX)); + + expect(refusal).toBeInstanceOf(ApproverDirectoryUnreadableError); + expect(refusal).toMatchObject({ approverType: 'queue', approverValue: 'billing' }); + expect(warnings.some(([msg]) => String(msg).includes("approver 'queue:billing' expanded to nobody"))).toBe(false); + nothingWritten(engine); + }); + + it('control — the readable slate opens as before', async () => { + const { svc } = outage([]); + const req = await svc.openNodeRequest(node([{ type: 'team', value: 't_ops' }, { type: 'position', value: 'cfo' }]), CTX) as any; + + expect(req.status).toBe('pending'); + expect([...req.pending_approvers].sort()).toEqual(['u5', 'u9']); + }); + + it('control — a department read that fails is not named by the ruling: the node still opens, as before', async () => { + const { engine, svc } = outage(['sys_business_unit', 'sys_business_unit_member']); + const outcome = await refusalOf(svc.openNodeRequest( + node([{ type: 'department', value: 'bu_x' }], { onEmptyApprovers: 'fallback', fallbackApprovers: [{ type: 'user', value: 'u_backstop' }] }), + CTX, + )); + + expect(outcome).toBeNull(); + expect(engine._tables['sys_approval_request']).toHaveLength(1); + }); + + it("decide's pre-snapshot re-resolution keeps the literal: a recorded decision is never failed by the outage", async () => { + // A unanimous request opened while the directory read, then stripped of + // its open-time snapshot as a request opened before the snapshot existed. + const failing: string[] = []; + const { engine, svc } = outage(failing); + const req = await svc.openNodeRequest( + node([{ type: 'user', value: 'u2' }, { type: 'team', value: 't_ops' }], { behavior: 'unanimous' }), CTX, + ) as any; + const row = engine._tables['sys_approval_request'].find((r: any) => r.id === req.id)!; + const config = JSON.parse(row.node_config_json); + delete config.__approverGroups; + row.node_config_json = JSON.stringify(config); + + failing.push('sys_team_member'); + const decided = await svc.decideNode(req.id, { decision: 'approve', actorId: 'u2' }, SYS); + expect(decided.decision).toBe('approve'); + }); +}); + +// [ADR-0142 D6 as amended; ruling record 6599731698 on #29604, letter B] The +// ADR-0105 D9 cross-organization read filter is a read behind a `position` or +// `team` approver as much as the directory expansion is. When it fails, the +// node refuses, naming THAT read, on the static resolver (its `bounded` +// filter) and on an `expression` approver's `resolveAs` values alike. +// `filterApproversWhoCanRead` answers its own failed `sys_member` read by +// keeping the slate, so the pass-through spy above is how it is made to fail. +// The scenario is `approver-cross-org.integration.test.ts`'s: a Plant A +// request routed to the group CFO. +describe('ApprovalService — a failed cross-organization read filter refuses the node, named as that read (ADR-0142 D6)', () => { + const PLANT_CTX = { userId: 'u_submitter', organizationId: 'o_plant', positions: [], permissions: [] } as any; + const FILTER_READ = /^the cross-organization read filter \(who among the people of organization 'o_group' can read this request of organization 'o_plant'\) could not be read: membership read filter down\.$/; + let engine: ReturnType; + let svc: ApprovalService; + let n = 0; + + beforeEach(() => { + vi.mocked(filterApproversWhoCanRead).mockClear(); + engine = makeFakeEngine(); + engine._tables['sys_user'].push({ id: 'u_cfo' }); + engine._tables['sys_organization'] = [ + { id: 'o_group', slug: 'acme-group', parent_organization_id: null }, + { id: 'o_plant', slug: 'acme-plant-a', parent_organization_id: 'o_group' }, + ]; + engine._tables['sys_user_position'] = [{ id: 'up1', user_id: 'u_cfo', position: 'cfo', organization_id: 'o_group' }]; + engine._tables['sys_member'] = [{ id: 'm1', user_id: 'u_cfo', organization_id: 'o_plant', role: 'member' }]; + n = 0; + svc = new ApprovalService({ + engine: engine as any, + clock: { now: () => new Date(1767000000000 + (n++) * 1000) }, + tenancyPosture: () => 'group', + }); + }); + + const node = (approver: Record) => ({ + object: 'purchase_order', + recordId: 'po1', + runId: 'run_1', + nodeId: 'group_signoff', + flowName: 'po_approval', + config: { approvers: [approver as any], behavior: 'unanimous' as const, lockRecord: false, onEmptyApprovers: 'admin_rescue' as const }, + record: { id: 'po1', signoff_position: 'cfo' }, + }); + const refusalOf = (p: Promise) => p.then(() => null, (e: unknown) => e); + + for (const [label, approver] of [ + ['a static position approver', { type: 'position', value: 'cfo', organization: '$root' }], + ['an expression approver resolving as position', { type: 'expression', value: 'trigger.signoff_position', resolveAs: 'position', organization: '$root' }], + ] as const) { + it(`control — ${label} reaches the group CFO through the filter`, async () => { + const req = await svc.openNodeRequest(node(approver), PLANT_CTX) as any; + expect(req.pending_approvers).toEqual(['u_cfo']); + expect(filterApproversWhoCanRead).toHaveBeenCalledTimes(1); + }); + + it(`${label}: the filter's failure refuses the node and names the filter's read`, async () => { + vi.mocked(filterApproversWhoCanRead).mockRejectedValueOnce(new Error('membership read filter down')); + const refusal = await refusalOf(svc.openNodeRequest(node(approver), PLANT_CTX)); + + expect(refusal).toBeInstanceOf(ApproverDirectoryUnreadableError); + expect(refusal).toMatchObject({ approverType: 'position', approverValue: 'cfo', retryable: true }); + expect((refusal as ApproverDirectoryUnreadableError).failedRead).toMatch(FILTER_READ); + expect(engine._tables['sys_approval_request'] ?? []).toEqual([]); + }); + } +}); + // #3807 follow-up: `queue` was not the only way to end up with a slot nobody // can act on — every GRAPH approver type falls back to the same literal when // its lookup finds no one, and that fallback used to happen in total silence. diff --git a/packages/plugins/plugin-approvals/src/approval-service.ts b/packages/plugins/plugin-approvals/src/approval-service.ts index 6fb82eb5804..b369ec2a059 100644 --- a/packages/plugins/plugin-approvals/src/approval-service.ts +++ b/packages/plugins/plugin-approvals/src/approval-service.ts @@ -1163,6 +1163,84 @@ function holderReadRefusal(scope: string, err: unknown): Error { ), { cause: err }); } +/** + * [ADR-0142 D6 as amended; ruling record 6599731698 on #29604, letter B] The + * approver types whose failed directory read refuses the approval node. A + * `queue`, `team` or `position` approver names a SET of people read from the + * directory at node entry; when that read fails, the slate it would build is + * missing people nobody can name, and every `onEmptyApprovers` policy would act + * on that wrong slate — open on the rest of it, open on the administrators or on + * a fallback, or approve. So the node opens on none of them. + * + * An `expression` approver's `resolveAs: 'team' | 'position'` values are read + * through the same expanders, and refuse the same way ({@link + * ApprovalService.resolveExpressionApprovers}). The other graph kinds + * (`department` and its dialects, `org_membership_level`) are not named by the + * ruling and keep their literal slot. + */ +const DIRECTORY_REFUSING_APPROVER_TYPES: ReadonlySet = new Set(['queue', 'team', 'position']); + +/** + * [#29604] The read a failed ADR-0105 D9 cross-organization read filter names: + * that filter's, not the directory's. One wording for the static resolver and + * the `expression` one, so a refusal from either says the same thing. + */ +function crossOrgFilterReadFailure(directoryOrg: unknown, organizationId: unknown, err: unknown): string { + return `the cross-organization read filter (who among the people of organization '${directoryOrg}' can read ` + + `this request of organization '${organizationId}') could not be read: ` + + `${err instanceof Error ? err.message : String(err)}.`; +} + +/** + * [ADR-0142 D6 as amended; ruling record 6599731698 on #29604, letter B] The + * approval node refuses to open because a directory read behind one of its + * approvers failed ({@link DIRECTORY_REFUSING_APPROVER_TYPES}). + * + * Its own class, and never `NO_APPROVERS` or "expanded to nobody": the slate + * was not found empty, it could not be read. The message names the approver + * and the read that failed, and starts with `APPROVER_DIRECTORY_UNREADABLE:` in + * the message-prefix family this service's refusals use. Retryable: nothing is + * written before it is thrown, so the run can be started again once the + * directory reads. No parked or auto-retry state is part of it. + */ +export class ApproverDirectoryUnreadableError extends Error { + override readonly name = 'ApproverDirectoryUnreadableError'; + /** The run can be re-triggered once the directory reads again. */ + readonly retryable = true; + /** The approver whose directory read failed: its type and value as authored. */ + readonly approverType: string; + readonly approverValue: string; + /** What could not be read, as the read reported it. */ + readonly failedRead: string; + + constructor(input: { + type: string; + value: unknown; + organizationId: string | null | undefined; + failedRead: string; + /** The CEL source, when the approver is an `expression` resolved as `type`. */ + expression?: string; + cause?: unknown; + }) { + const approver = input.expression === undefined + ? `approver '${input.type}:${input.value}'` + : `approver 'expression' (resolveAs '${input.type}', value '${input.value}')`; + const inOrg = input.organizationId ? ` in organization '${input.organizationId}'` : ''; + const read = input.failedRead.trim(); + super( + `APPROVER_DIRECTORY_UNREADABLE: ${approver} could not be expanded${inOrg}, because a read behind it failed: ` + + `${read}${/[.!?]$/.test(read) ? '' : '.'} The approval node does not open: the people who could not be ` + + `read would be routed past, so the request is not opened on the rest of the slate, on the declared ` + + `fallbackApprovers or on the organization's administrators, and is not auto-approved, whatever the node's ` + + `onEmptyApprovers says. Run the flow again once the directory can be read.`, + ); + if (input.cause !== undefined) Object.assign(this, { cause: input.cause }); + this.approverType = input.type; + this.approverValue = String(input.value); + this.failedRead = input.failedRead; + } +} + /** * Approver types resolved by QUERYING a graph rather than by taking `value` * literally (#3807). Each can legitimately come back empty — an unstaffed @@ -2432,9 +2510,19 @@ export class ApprovalService implements IApprovalService { * as `__resolvedFrom` so "why these people" stays answerable later. */ resolvedFrom?: Record; + /** + * [ADR-0142 D6 as amended] What a failed directory read behind a + * `queue` / `team` / `position` approver does: `'refuse'` (the default) + * throws {@link ApproverDirectoryUnreadableError}, so a node never opens on + * a slate it could not read; `'literal'` keeps the slot's `type:value` + * literal, for the one reader that re-resolves a slate AFTER a decision + * was recorded and must not fail it (`decide`'s pre-snapshot tally). + */ + onDirectoryFailure?: 'refuse' | 'literal'; }, ): Promise { if (!step || !Array.isArray(step.approvers)) return []; + const onDirectoryFailure = opts?.onDirectoryFailure ?? 'refuse'; const now = opts?.now ?? this.clock.now().getTime(); const out: string[] = []; // A person is a member of a per_group group once (#28684): the tally sizes @@ -2468,7 +2556,7 @@ export class ApprovalService implements IApprovalService { // contract of resolveApproverSpec cannot carry. if (canonicalApproverType(String(a.type)) === 'expression') { const resolved = await this.resolveExpressionApprovers( - a, record, organizationId, now, opts?.substitutions, opts?.exprCtx, + a, record, organizationId, now, opts?.substitutions, opts?.exprCtx, onDirectoryFailure, ); if (opts?.resolvedFrom) opts.resolvedFrom[`expression#${idx}`] = resolved.raw; for (const entry of resolved.slots) { @@ -2484,7 +2572,7 @@ export class ApprovalService implements IApprovalService { if (opts?.resolvedFrom && canonicalApproverType(String(a.type)) === 'field' && a.value != null) { opts.resolvedFrom[`field:${a.value}`] = (record as any)?.[a.value] ?? null; } - const ids = await this.resolveApproverSpec(a, record, organizationId, now, opts?.substitutions); + const ids = await this.resolveApproverSpec(a, record, organizationId, now, opts?.substitutions, onDirectoryFailure); // per_group (#3266): tag each resolved id with this spec's group. for (const u of ids) { if (!u) continue; @@ -2513,6 +2601,7 @@ export class ApprovalService implements IApprovalService { organizationId: string | null | undefined, now: number, substitutions?: OooSubstitution[], + onDirectoryFailure: 'refuse' | 'literal' = 'refuse', ): Promise { // ADR-0090 D3: `role` is the deprecated spelling of `org_membership_level`. // Resolve on the canonical type, and write the canonical type in the @@ -2602,14 +2691,21 @@ export class ApprovalService implements IApprovalService { // // A cross-org slate is filtered to the people who can actually READ the // request (D2 union); same-org routing is untouched and does no extra read. - const bounded = async (users: string[]): Promise => ( - crossOrg - ? filterApproversWhoCanRead(this.orgScopeDeps, users, organizationId, { + // + // [#29604] A failure of the read filter itself is named as that read, not + // as the directory's, so the warning and the node's refusal below say which + // read could not be done. + const bounded = async (users: string[]): Promise => { + if (!crossOrg) return users; + try { + return await filterApproversWhoCanRead(this.orgScopeDeps, users, organizationId, { approverType: type, value: a.value != null ? String(a.value) : undefined, directoryOrgId: directoryOrg, - }) - : users - ); + }); + } catch (err) { + throw Object.assign(new Error(crossOrgFilterReadFailure(directoryOrg, organizationId, err)), { cause: err }); + } + }; // The user a `manager` approver resolved FROM, kept for the warning below // when no manager comes back. @@ -2659,6 +2755,17 @@ export class ApprovalService implements IApprovalService { // A directory lookup failed → handled below as resolving to nobody, and // the graph-type warning names the failure (#29362). directoryFailure = err instanceof Error ? err.message : String(err); + // [ADR-0142 D6 as amended; ruling record 6599731698, letter B] …unless + // the approver names a directory SET: then the node refuses to open, + // before any `onEmptyApprovers` policy can act on a slate missing the + // people who could not be read. Ahead of the `queue` branch's "expanded to + // nobody" return below, which would otherwise answer an unreadable queue + // as an empty one. + if (onDirectoryFailure === 'refuse' && DIRECTORY_REFUSING_APPROVER_TYPES.has(type)) { + throw new ApproverDirectoryUnreadableError({ + type, value: a.value, organizationId: directoryOrg, failedRead: directoryFailure, cause: err, + }); + } } // #22558: a PERSON type that resolved to nobody contributes NO slot — see // PERSON_APPROVER_TYPES for why its `type:value` literal could never be @@ -2695,7 +2802,19 @@ export class ApprovalService implements IApprovalService { // are read at node entry, so a member who joins later is not owed this // request. A `queue:` slot a stored request already holds stays as stored, // decidable by the override (the ADR-0090 D3 `role:` precedent). + // + // A queue whose read FAILED reaches here only on the `'literal'` reader (a + // refusing one threw above); it still writes no slot, and its warning says + // the read failed rather than that the queue is empty. if (type === 'queue') { + if (directoryFailure !== undefined) { + this.logger?.warn?.( + `[approvals] approver 'queue:${a.value}' could not be expanded — a directory or read-filter lookup failed. ` + + `It adds no slot to this slate, as an empty queue adds none; fix the read (see error).`, + { type, value: a.value ?? null, organizationId: directoryOrg ?? null, error: directoryFailure }, + ); + return []; + } this.logger?.warn?.( `[approvals] approver 'queue:${a.value}' expanded to nobody — ` + (this.queueSource() @@ -2797,6 +2916,7 @@ export class ApprovalService implements IApprovalService { now: number, substitutions?: OooSubstitution[], exprCtx?: ApproverExpressionContext, + onDirectoryFailure: 'refuse' | 'literal' = 'refuse', ): Promise<{ slots: Array<{ id: string; subGroup?: string }>; raw: string[] }> { const source = String(a.value ?? '').trim(); if (!source) { @@ -2916,6 +3036,15 @@ export class ApprovalService implements IApprovalService { } } catch (err: any) { if (String(err?.message ?? '').startsWith('VALIDATION_FAILED')) throw err; + // [ADR-0142 D6 as amended; ruling record 6599731698, letter B] A + // `team` or `position` value is read through the same expanders as the + // static types, and refuses the node the same way. + if (onDirectoryFailure === 'refuse' && DIRECTORY_REFUSING_APPROVER_TYPES.has(resolveAs)) { + throw new ApproverDirectoryUnreadableError({ + type: resolveAs, value: key, organizationId: directoryOrg, + failedRead: err instanceof Error ? err.message : String(err), expression: source, cause: err, + }); + } // [#29606] The literal slot below is the same fallback the static graph // types take, and it says why exactly as they do: a failed read is not // an unstaffed value, so an operator is pointed at the read rather than @@ -2924,9 +3053,23 @@ export class ApprovalService implements IApprovalService { users = []; } if (crossOrg && users.length) { - users = await filterApproversWhoCanRead(this.orgScopeDeps, users, organizationId, { - approverType: resolveAs, value: key, directoryOrgId: directoryOrg, - }); + try { + users = await filterApproversWhoCanRead(this.orgScopeDeps, users, organizationId, { + approverType: resolveAs, value: key, directoryOrgId: directoryOrg, + }); + } catch (err) { + // [#29604] The read filter is a read behind a `team` or `position` + // value as much as the expansion is: it refuses the node the same + // way, naming the filter's read as the static resolver's `bounded` + // does. Every other case fails as it always did. + if (onDirectoryFailure === 'refuse' && DIRECTORY_REFUSING_APPROVER_TYPES.has(resolveAs)) { + throw new ApproverDirectoryUnreadableError({ + type: resolveAs, value: key, organizationId: directoryOrg, + failedRead: crossOrgFilterReadFailure(directoryOrg, organizationId, err), expression: source, cause: err, + }); + } + throw err; + } } if (!users.length) { slots.push({ id: `${resolveAs}:${key}`, subGroup: key }); @@ -4985,8 +5128,12 @@ export class ApprovalService implements IApprovalService { groupMap = snapshotGroups; original = Object.keys(snapshotGroups); } else { + // The decision is already recorded above, so a directory outage here + // must not fail it: the legacy re-resolution keeps the literal slot it + // always kept (ADR-0142 D6's refusal is the node's, at open time). original = await this.expandApprovers( { approvers: config.approvers }, parseJson(raw.payload_json, undefined), org, + { onDirectoryFailure: 'literal' }, ); groupMap = {}; } diff --git a/packages/plugins/plugin-approvals/src/business-unit-holders-complete.integration.test.ts b/packages/plugins/plugin-approvals/src/business-unit-holders-complete.integration.test.ts index 45da10bbfd0..b8896b9dd88 100644 --- a/packages/plugins/plugin-approvals/src/business-unit-holders-complete.integration.test.ts +++ b/packages/plugins/plugin-approvals/src/business-unit-holders-complete.integration.test.ts @@ -30,7 +30,7 @@ import { SqlDriver } from '@objectstack/driver-sql'; import { SysBusinessUnit, SysBusinessUnitMember, SysMember, SysOrganization, SysTeamMember, } from '@objectstack/platform-objects/identity'; -import { ApprovalService } from './approval-service.js'; +import { ApprovalService, ApproverDirectoryUnreadableError } from './approval-service.js'; const ORG = 'org_large'; const ELSEWHERE = 'org_elsewhere'; @@ -129,8 +129,8 @@ const department = (rig: Rig, value = ROOT) => (rig.svc as any).expandBusinessUnitUsers(value, ORG) as Promise; const resolveSpec = (rig: Rig, a: { type: string; value: string }) => (rig.svc as any).resolveApproverSpec(a, undefined, ORG, Date.now()) as Promise; -const resolveExpression = (rig: Rig, a: { type: string; value: string; resolveAs: string }) => - (rig.svc as any).resolveExpressionApprovers(a, {}, ORG, Date.now()) as Promise<{ +const resolveExpression = (rig: Rig, a: { type: string; value: string; resolveAs: string }, onDirectoryFailure?: 'refuse' | 'literal') => + (rig.svc as any).resolveExpressionApprovers(a, {}, ORG, Date.now(), undefined, undefined, onDirectoryFailure) as Promise<{ slots: Array<{ id: string; subGroup?: string }>; raw: string[]; }>; @@ -220,10 +220,18 @@ describe('[#29606] item 3 — an expression approver whose resolveAs expansion f expect(failed[0][1].error).toMatch(/SQLITE_IOERR/); }); - it('⛔ a failed team read through resolveAs warns the same way', async () => { + it('⛔ a failed team read through resolveAs refuses the node as a team approver does; the literal reader warns the same way', async () => { const rig = await boot(0); rig.failReadsOf('sys_team_member'); - const { slots } = await resolveExpression(rig, { type: 'expression', value: `'team_x'`, resolveAs: 'team' }); + // [ADR-0142 D6 as amended, #29604] The value is read through the team + // expander, so its failure refuses the node like a static `team` approver's. + const refusal = await resolveExpression(rig, { type: 'expression', value: `'team_x'`, resolveAs: 'team' }) + .then(() => null, (e: unknown) => e); + expect(refusal).toBeInstanceOf(ApproverDirectoryUnreadableError); + expect(refusal).toMatchObject({ approverType: 'team', approverValue: 'team_x', retryable: true }); + expect((refusal as Error).message).toMatch(/^APPROVER_DIRECTORY_UNREADABLE: approver 'expression' \(resolveAs 'team', value 'team_x'\)/); + + const { slots } = await resolveExpression(rig, { type: 'expression', value: `'team_x'`, resolveAs: 'team' }, 'literal'); expect(slots).toEqual([{ id: 'team:team_x', subGroup: 'team_x' }]); const failed = failureWarnings(rig); diff --git a/packages/plugins/plugin-approvals/src/member-tier-holders-complete.integration.test.ts b/packages/plugins/plugin-approvals/src/member-tier-holders-complete.integration.test.ts index 47d16e816ef..a73e1788d5e 100644 --- a/packages/plugins/plugin-approvals/src/member-tier-holders-complete.integration.test.ts +++ b/packages/plugins/plugin-approvals/src/member-tier-holders-complete.integration.test.ts @@ -17,12 +17,15 @@ * and an `org_membership_level` approver logged "expanded to nobody", * indistinguishable from a tier nobody holds. * - * What a caller sees now: a failed read throws out of the expansion, the - * approver-spec resolver's directory catch takes it, and the slot falls back - * to its literal (`position:

`, acted on by whoever holds the position at - * decision time; `org_membership_level:`, decided by the override) with a - * warning that names the failed read and carries its `error` — never a slate - * cut down to the part that happened to read. + * What a caller sees now: a failed read throws out of the expansion and the + * approver-spec resolver's directory catch takes it. An `org_membership_level` + * slot falls back to its literal (`org_membership_level:`, decided by the + * override) with a warning that names the failed read and carries its `error`. + * [ADR-0142 D6 as amended, #29604] A `position` approver, whose transition limb + * this read is, refuses the approval node instead + * (`ApproverDirectoryUnreadableError`); only the `'literal'` reader keeps its + * `position:

` literal. Never a slate cut down to the part that happened to + * read. * * Measured on a REAL engine (ObjectQL + SqlDriver over in-memory SQLite) * carrying the shipped `SysMember`, because the completion is a keyset read @@ -33,7 +36,7 @@ import { describe, it, expect, afterEach, vi } from 'vitest'; import { ObjectQL } from '@objectstack/objectql'; import { SqlDriver } from '@objectstack/driver-sql'; import { SysMember, SysOrganization } from '@objectstack/platform-objects/identity'; -import { ApprovalService } from './approval-service.js'; +import { ApprovalService, ApproverDirectoryUnreadableError } from './approval-service.js'; const ORG = 'org_large'; const ELSEWHERE = 'org_elsewhere'; @@ -112,8 +115,8 @@ async function boot(n: number, opts: { registerMember?: boolean } = {}): Promise /** The private expansions, as the approver-spec resolver calls them. */ const tier = (rig: Rig, value: string) => (rig.svc as any).expandMembershipTierUsers(value, ORG) as Promise; const position = (rig: Rig, value: string) => (rig.svc as any).expandPositionUsers(value, ORG) as Promise; -const resolveSpec = (rig: Rig, a: { type: string; value: string }) => - (rig.svc as any).resolveApproverSpec(a, undefined, ORG, Date.now()) as Promise; +const resolveSpec = (rig: Rig, a: { type: string; value: string }, onDirectoryFailure?: 'refuse' | 'literal') => + (rig.svc as any).resolveApproverSpec(a, undefined, ORG, Date.now(), undefined, onDirectoryFailure) as Promise; const failMemberReads = (rig: Rig, fault: Error) => { const memberDriver = (rig.engine as any).getDriverForObject('sys_member'); @@ -168,11 +171,18 @@ describe('[#29362] approvals membership-tier expansion reads every holder (drive expect(rig.warn.mock.calls.some(([, meta]) => typeof meta?.error === 'string')).toBe(false); }); - it('⛔ what the caller sees: the slot falls back to its literal, and the warning names the failed read', async () => { + it('⛔ what the caller sees: a position refuses the node on its sys_member limb; the tier slot keeps its literal', async () => { const rig = await boot(3); failMemberReads(rig, new Error('SQLITE_IOERR: disk I/O error')); - expect(await resolveSpec(rig, { type: 'position', value: TIER })).toEqual([`position:${TIER}`]); + // [#29604] The ADR-0057 D4 transition limb is a read behind the position: + // its failure is a failed read, and the node refuses on it. + const refusal = await resolveSpec(rig, { type: 'position', value: TIER }).then(() => null, (e: unknown) => e); + expect(refusal).toBeInstanceOf(ApproverDirectoryUnreadableError); + expect(refusal).toMatchObject({ approverType: 'position', approverValue: TIER, retryable: true }); + expect((refusal as ApproverDirectoryUnreadableError).failedRead).toMatch(/SQLITE_IOERR/); + + expect(await resolveSpec(rig, { type: 'position', value: TIER }, 'literal')).toEqual([`position:${TIER}`]); expect(await resolveSpec(rig, { type: 'org_membership_level', value: TIER })).toEqual([`org_membership_level:${TIER}`]); const failed = rig.warn.mock.calls.filter(([, meta]) => typeof meta?.error === 'string'); expect(failed.map(([, meta]) => meta.type)).toEqual(['position', 'org_membership_level']); diff --git a/packages/plugins/plugin-approvals/src/team-position-holders-complete.integration.test.ts b/packages/plugins/plugin-approvals/src/team-position-holders-complete.integration.test.ts index a1374a3cc7b..aa86ee56e00 100644 --- a/packages/plugins/plugin-approvals/src/team-position-holders-complete.integration.test.ts +++ b/packages/plugins/plugin-approvals/src/team-position-holders-complete.integration.test.ts @@ -16,10 +16,13 @@ * - a failed `sys_user_position` read routed a `position` approver to its * `sys_member.role` holders ALONE — a shrunk slate that looks complete. * - * What a caller sees now: the expansion throws, the approver-spec resolver's - * directory catch takes it, and the slot falls back to its literal with the - * warning that names the failed read and carries its `error`. A membership - * object that does not exist is still no holders. + * What a caller sees now: the expansion throws and the approver-spec + * resolver's directory catch takes it. [ADR-0142 D6 as amended, #29604] The + * resolver then refuses the approval node (`ApproverDirectoryUnreadableError`, + * naming the read), so no `onEmptyApprovers` policy acts on the slate; only + * `decide`'s pre-snapshot re-resolution (the `'literal'` reader) still falls + * back to the literal, with the warning that names the failed read and carries + * its `error`. A membership object that does not exist is still no holders. * * Measured on a REAL engine (ObjectQL + SqlDriver over in-memory SQLite) * carrying the shipped `SysTeamMember` and `SysMember`; `sys_user_position` is @@ -31,7 +34,7 @@ import { ObjectQL } from '@objectstack/objectql'; import { SqlDriver } from '@objectstack/driver-sql'; import { ObjectSchema, Field } from '@objectstack/spec/data'; import { SysMember, SysOrganization, SysTeamMember } from '@objectstack/platform-objects/identity'; -import { ApprovalService } from './approval-service.js'; +import { ApprovalService, ApproverDirectoryUnreadableError } from './approval-service.js'; const ORG = 'org_large'; const ELSEWHERE = 'org_elsewhere'; @@ -137,8 +140,8 @@ async function boot(seeded: 'team' | 'position', n: number, opts: { omit?: strin /** The private expansions and the resolver, as the request opener calls them. */ const team = (rig: Rig, value = TEAM) => (rig.svc as any).expandTeamUsers(value, ORG) as Promise; const position = (rig: Rig, value = POSITION) => (rig.svc as any).expandPositionUsers(value, ORG) as Promise; -const resolveSpec = (rig: Rig, a: { type: string; value: string }) => - (rig.svc as any).resolveApproverSpec(a, undefined, ORG, Date.now()) as Promise; +const resolveSpec = (rig: Rig, a: { type: string; value: string }, onDirectoryFailure?: 'refuse' | 'literal') => + (rig.svc as any).resolveApproverSpec(a, undefined, ORG, Date.now(), undefined, onDirectoryFailure) as Promise; const failureWarnings = (rig: Rig) => rig.warn.mock.calls.filter(([, meta]) => typeof meta?.error === 'string'); describe('[#29503] approvals team expansion reads every member (driver-sql, shipped SysTeamMember)', () => { @@ -173,11 +176,18 @@ describe('[#29503] approvals team expansion reads every member (driver-sql, ship await expect(team(rig)).rejects.toThrow(/members of team 'team_large'.*could not be read/); }); - it('⛔ what the caller sees: the team slot falls back to its literal, and the warning names the failed read', async () => { + it('⛔ what the caller sees: the node refuses, naming the failed read; the literal reader keeps the literal and warns', async () => { const rig = await boot('team', 3); rig.failReadsOf('sys_team_member'); - expect(await resolveSpec(rig, { type: 'team', value: TEAM })).toEqual([`team:${TEAM}`]); + const refusal = await resolveSpec(rig, { type: 'team', value: TEAM }).then(() => null, (e: unknown) => e); + expect(refusal).toBeInstanceOf(ApproverDirectoryUnreadableError); + expect(refusal).toMatchObject({ approverType: 'team', approverValue: TEAM, retryable: true }); + expect((refusal as Error).message).toMatch(/^APPROVER_DIRECTORY_UNREADABLE: approver 'team:team_large'/); + expect((refusal as ApproverDirectoryUnreadableError).failedRead).toMatch(/members of team 'team_large'.*could not be read/); + expect(failureWarnings(rig)).toEqual([]); + + expect(await resolveSpec(rig, { type: 'team', value: TEAM }, 'literal')).toEqual([`team:${TEAM}`]); const failed = failureWarnings(rig); expect(failed.map(([, meta]) => meta.type)).toEqual(['team']); expect(failed[0][0]).toMatch(/could not be expanded — a directory or read-filter lookup failed/); @@ -221,14 +231,19 @@ describe('[#29503] approvals position expansion reads every assignment (driver-s await expect(position(rig)).rejects.toThrow(/assignments of position 'cfo_office'.*could not be read/); }); - it('⛔ what the caller sees: the position slot falls back to its literal, never the tier holders alone', async () => { + it('⛔ what the caller sees: the node refuses, never the tier holders alone; the literal reader keeps the literal', async () => { const rig = await boot('position', 3); await (rig.engine as any).getDriverForObject('sys_member').bulkCreate('sys_member', [ { id: 'mem_tier', user_id: 'usr_tier', organization_id: ORG, role: POSITION }, ]); rig.failReadsOf('sys_user_position'); - expect(await resolveSpec(rig, { type: 'position', value: POSITION })).toEqual([`position:${POSITION}`]); + const refusal = await resolveSpec(rig, { type: 'position', value: POSITION }).then(() => null, (e: unknown) => e); + expect(refusal).toBeInstanceOf(ApproverDirectoryUnreadableError); + expect(refusal).toMatchObject({ approverType: 'position', approverValue: POSITION, retryable: true }); + expect((refusal as ApproverDirectoryUnreadableError).failedRead).toMatch(/assignments of position 'cfo_office'.*could not be read/); + + expect(await resolveSpec(rig, { type: 'position', value: POSITION }, 'literal')).toEqual([`position:${POSITION}`]); const failed = failureWarnings(rig); expect(failed.map(([, meta]) => meta.type)).toEqual(['position']); expect(failed[0][1].error).toMatch(/SQLITE_IOERR/); diff --git a/packages/plugins/plugin-sharing/src/ownership-queue-service.ts b/packages/plugins/plugin-sharing/src/ownership-queue-service.ts index 2fad5375487..8f03d2573a2 100644 --- a/packages/plugins/plugin-sharing/src/ownership-queue-service.ts +++ b/packages/plugins/plugin-sharing/src/ownership-queue-service.ts @@ -37,6 +37,18 @@ * read of the activation ledger leaves the declared default — on — for that * call and is logged, the posture the sharing-rule service takes for the same * ledger. + * + * ## Read to completion + * + * [#29604] The authored reads — an organization's `sys_queue` rows and its + * `sys_queue_member` rows — are read to completion through the keyset walk the + * team, position and business-unit member reads share + * (`readMembershipToCompletion`), never cut at a row bound: a queue past the + * bound lost its later members, a member past it lost every queue it sits in, + * and nothing said so. A read that fails, or a walk that cannot vouch for + * completion, takes the failed-read answer above. ADR-0142 D2 as amended has + * the recompute consumers (D6, D7) refuse a failed read instead; that answer + * waits on the contract carrying the caller's call-time signal. */ import type { @@ -49,7 +61,7 @@ import { QueueMemberType } from '@objectstack/spec/security'; import { METADATA_ACTIVATION_TABLE } from '@objectstack/core'; import type { OptionalSharingLogger } from './logger-shapes.js'; import type { SharingEngine } from './sharing-service.js'; -import { TeamGraphService, teamHasMember } from './team-graph.js'; +import { TeamGraphService, readMembershipToCompletion, teamHasMember } from './team-graph.js'; import { PositionGraphService } from './position-graph.js'; import { BusinessUnitGraphService } from './business-unit-graph.js'; @@ -62,9 +74,6 @@ export const QUEUE_MEMBER_OBJECT = 'sys_queue_member'; /** The `sys_metadata_activation.metadata_type` a declared queue is switched by. */ export const QUEUE_ACTIVATION_TYPE = 'queue'; -/** Upper bound on one organization's authored queues / member rows read in one call. */ -const QUEUE_READ_LIMIT = 10000; - /** The member kinds a specification may carry, as a set for the row guard. */ const MEMBER_KINDS: ReadonlySet = new Set(QueueMemberType.options); @@ -108,6 +117,21 @@ function isRecord(v: unknown): v is Record { return v !== null && typeof v === 'object' && !Array.isArray(v); } +/** + * The reason behind a failed authored read, for a warning that answers nobody. + * A walk's refusal ends on its own consequence ("…, so the expansion refuses + * instead."), which is not this service's answer until #29604 part 2, so only + * the reason is logged: the driver's error the refusal carries as `cause`, or, + * for a walk that could not vouch for completion (no cause), the refusal's + * sentence with that closing clause cut. + */ +function readFailureReason(err: unknown): string { + const cause = (err as { cause?: unknown } | null)?.cause; + if (cause instanceof Error) return cause.message; + const message = err instanceof Error ? err.message : String(err); + return message.replace(/ [^.]*\brefuses instead\.$/, ''); +} + /** * One call's memo of the reads several specifications share: a position's * activation and holding, a team's membership, a unit's expansion. Scoped to a @@ -268,22 +292,22 @@ export class OwnershipQueueService implements IOwnershipQueueService { } } try { - const rows = await engine.find(QUEUE_MEMBER_OBJECT, { + const rows = await readMembershipToCompletion(engine, { + object: QUEUE_MEMBER_OBJECT, where: { recipient_type: kind, ...(organizationId ? { organization_id: organizationId } : {}), }, fields: ['queue', 'recipient_id'], - limit: QUEUE_READ_LIMIT, - context: SYSTEM_CTX, + subject: `the '${kind}' member rows of the queues${organizationId ? ` in organization '${organizationId}'` : ''}`, }); - for (const r of (Array.isArray(rows) ? rows : []) as Array>) { + for (const r of rows ?? []) { const queue = str(r.queue); const value = str(r.recipient_id); if (queue && value && (BU_MEMBER_KINDS.has(kind) || values.has(value))) out.add(queue); } } catch (err: any) { - this.warnOnce(`naming::${kind}`, '[queue] member read failed while looking through queues — the rules naming those queues are re-evaluated on their next write', { kind, error: err?.message }); + this.warnOnce(`naming::${kind}`, '[queue] member read failed while looking through queues — the rules naming those queues are re-evaluated on their next write', { kind, error: readFailureReason(err) }); } return [...out].sort(); } @@ -388,8 +412,9 @@ export class OwnershipQueueService implements IOwnershipQueueService { * * The checks themselves (`teamHasMember`, `PositionGraphService.holdsPosition`) * now refuse a failed read instead of answering `false`; this keeps the - * queue's consumers where they were. Whether a queue should refuse too is not - * decided by ADR-0142 and is left to a ruling rather than narrowed here. + * queue's consumers where they were. ADR-0142 D2 as amended (#29604) has the + * recompute consumers refuse instead; the call-time signal they say so with + * is the contract's to carry, and until it does every caller gets this answer. */ private async holderCheck(spec: QueueMemberSpec, check: () => Promise): Promise { try { @@ -442,14 +467,14 @@ export class OwnershipQueueService implements IOwnershipQueueService { declaredNames: ReadonlySet, ): Promise { try { - const rows = await engine.find(QUEUE_OBJECT, { + const rows = await readMembershipToCompletion(engine, { + object: QUEUE_OBJECT, where: organizationId ? { organization_id: organizationId } : {}, fields: ['name'], - limit: QUEUE_READ_LIMIT, - context: SYSTEM_CTX, + subject: `the authored queues${organizationId ? ` of organization '${organizationId}'` : ''}`, }); const out: string[] = []; - for (const r of (Array.isArray(rows) ? rows : []) as Array>) { + for (const r of rows ?? []) { const name = str(r.name); // A row under a declared name is not a second queue (the guard refuses // the write; a row that predates the declaration is reported there). @@ -457,7 +482,7 @@ export class OwnershipQueueService implements IOwnershipQueueService { } return out; } catch (err: any) { - this.warnOnce('read::sys_queue', '[queue] sys_queue read failed — authored queues resolve nowhere until it succeeds', { error: err?.message }); + this.warnOnce('read::sys_queue', '[queue] sys_queue read failed — authored queues resolve nowhere until it succeeds', { error: readFailureReason(err) }); return []; } } @@ -471,16 +496,16 @@ export class OwnershipQueueService implements IOwnershipQueueService { const out = new Map(); if (names.length === 0) return out; try { - const rows = await engine.find(QUEUE_MEMBER_OBJECT, { + const rows = await readMembershipToCompletion(engine, { + object: QUEUE_MEMBER_OBJECT, where: { queue: names.length === 1 ? names[0] : { $in: [...names] }, ...(organizationId ? { organization_id: organizationId } : {}), }, fields: ['queue', 'recipient_type', 'recipient_id'], - limit: QUEUE_READ_LIMIT, - context: SYSTEM_CTX, + subject: `the member rows of ${names.length === 1 ? `queue '${names[0]}'` : `${names.length} queues`}${organizationId ? ` in organization '${organizationId}'` : ''}`, }); - for (const r of (Array.isArray(rows) ? rows : []) as Array>) { + for (const r of rows ?? []) { const queue = str(r.queue); const type = str(r.recipient_type); const value = str(r.recipient_id); @@ -490,7 +515,7 @@ export class OwnershipQueueService implements IOwnershipQueueService { out.set(queue, list); } } catch (err: any) { - this.warnOnce('read::sys_queue_member', '[queue] sys_queue_member read failed — authored members count as nobody until it succeeds', { error: err?.message }); + this.warnOnce('read::sys_queue_member', '[queue] sys_queue_member read failed — authored members count as nobody until it succeeds', { error: readFailureReason(err) }); } return out; } diff --git a/packages/plugins/plugin-sharing/src/ownership-queue.test.ts b/packages/plugins/plugin-sharing/src/ownership-queue.test.ts index 5370315e3f6..f5d8402d6cb 100644 --- a/packages/plugins/plugin-sharing/src/ownership-queue.test.ts +++ b/packages/plugins/plugin-sharing/src/ownership-queue.test.ts @@ -25,8 +25,9 @@ import { ruleRegrantQueue } from './rule-hooks.js'; type Row = Record; /** - * Equality, `$in`, `$or` and `$and` — every predicate shape the queue reads - * send. Any other combinator or operator is refused rather than misread. + * Equality, `$in`, `$gt`, `$or` and `$and` — every predicate shape the queue + * reads send (`$gt` is the keyset cursor of a paged read, #29604). Any other + * combinator or operator is refused rather than misread. */ function matches(row: Row, filter: any): boolean { if (!filter || typeof filter !== 'object') return true; @@ -43,6 +44,10 @@ function matches(row: Row, filter: any): boolean { const rv = row[k] ?? null; if (v != null && typeof v === 'object') { const ops = Object.keys(v as object); + if (ops.length === 1 && ops[0] === '$gt') { + if (!(rv != null && String(rv) > String((v as any).$gt))) return false; + continue; + } if (ops.length !== 1 || ops[0] !== '$in') throw new Error(`test double: operator ${ops.join(',')} on '${k}' is not implemented`); if (!(v as any).$in.includes(rv)) return false; continue; @@ -72,6 +77,9 @@ function makeEngine(opts: { if (opts.failOn?.includes(object)) throw new Error(`${object} unavailable`); const filter = options?.filter ?? options?.where; const rows = (tables[object] ?? []).filter((r) => matches(r, filter)); + for (const { field } of [...(options?.orderBy ?? [])].reverse()) { + rows.sort((a, b) => (String(a[field]) < String(b[field]) ? -1 : String(a[field]) > String(b[field]) ? 1 : 0)); + } return typeof options?.limit === 'number' ? rows.slice(0, options.limit) : rows; }, async insert(object: string, data: any) { @@ -117,11 +125,12 @@ const BASE_TABLES = (): Record => ({ { id: 'q1', name: 'billing', label: 'Billing', organization_id: 'org_a' }, { id: 'q2', name: 'billing', label: 'Billing B', organization_id: 'org_b' }, ], + // [#29604] Carrying their `id` too: the authored queue reads page by it. sys_queue_member: [ - { queue: 'billing', recipient_type: 'position', recipient_id: 'billing_clerk', organization_id: 'org_a' }, - { queue: 'billing', recipient_type: 'user', recipient_id: 'u_e', organization_id: 'org_a' }, - { queue: 'support_tier1', recipient_type: 'team', recipient_id: 't_night', organization_id: 'org_a' }, - { queue: 'billing', recipient_type: 'user', recipient_id: 'u_b_only', organization_id: 'org_b' }, + { id: 'qm_1', queue: 'billing', recipient_type: 'position', recipient_id: 'billing_clerk', organization_id: 'org_a' }, + { id: 'qm_2', queue: 'billing', recipient_type: 'user', recipient_id: 'u_e', organization_id: 'org_a' }, + { id: 'qm_3', queue: 'support_tier1', recipient_type: 'team', recipient_id: 't_night', organization_id: 'org_a' }, + { id: 'qm_4', queue: 'billing', recipient_type: 'user', recipient_id: 'u_b_only', organization_id: 'org_b' }, ], [METADATA_ACTIVATION_TABLE]: [], }); @@ -235,7 +244,7 @@ describe('[ADR-0142 D2] OwnershipQueueService — the one expander', () => { it('queuesOf lists every queue the user is in, declared and authored, ordered by name', async () => { const tables = BASE_TABLES(); - tables.sys_queue_member.push({ queue: 'billing', recipient_type: 'user', recipient_id: 'u_a', organization_id: 'org_a' }); + tables.sys_queue_member.push({ id: 'qm_5', queue: 'billing', recipient_type: 'user', recipient_id: 'u_a', organization_id: 'org_a' }); const svc = service(makeEngine({ declared: SUPPORT_DECLARED, tables })); expect(await svc.queuesOf('u_a', 'org_a')).toEqual(['billing', 'support_tier1']); expect(await svc.queuesOf('u_c', 'org_a')).toEqual(['billing']); @@ -293,6 +302,91 @@ describe('[ADR-0142 D2] OwnershipQueueService — the one expander', () => { // ─── "Does this organization have any queue at all?" ─────────────────────── +// [#29604, review record 6591670782] The queue reads paged by id to completion, +// where they read one page of 10,000 rows and answered on it: a queue's +// 10,001st member, an organization's 10,001st queue, and the queues naming a +// recipient past that bound were silently not there. A read that fails partway +// keeps the answer a failed read gives today (nobody, logged once) until the +// contract can carry a refusal (#30251) — never the first page as if complete. +describe('[ADR-0142 D2] OwnershipQueueService — the queue reads page to completion (#29604)', () => { + const PAST_ONE_PAGE = 10_001; + const pad = (i: number) => String(i).padStart(5, '0'); + const LAST = pad(PAST_ONE_PAGE - 1); + + /** `count` user members of queue `big` in org_a, keyed so the last sorts last. */ + const bigQueue = (count = PAST_ONE_PAGE): Record => ({ + ...BASE_TABLES(), + sys_queue: [{ id: 'q_big', name: 'big', label: 'Big', organization_id: 'org_a' }], + sys_queue_member: Array.from({ length: count }, (_, i) => ({ + id: `qm_${pad(i)}`, queue: 'big', recipient_type: 'user', recipient_id: `u_${pad(i)}`, organization_id: 'org_a', + })), + }); + + it('a queue past one page expands to every member, and its last member is one', async () => { + const svc = service(makeEngine({ tables: bigQueue() })); + const members = await svc.expandMembers('big', 'org_a'); + expect(members).toHaveLength(PAST_ONE_PAGE); + expect(members.at(-1)).toBe(`u_${LAST}`); + expect(await svc.isMember(`u_${LAST}`, 'big', 'org_a')).toBe(true); + expect(await svc.queuesOf(`u_${LAST}`, 'org_a')).toEqual(['big']); + }); + + it('an organization past one page of queues lists its last queue for that queue\'s member', async () => { + const tables = BASE_TABLES(); + tables.sys_queue = Array.from({ length: PAST_ONE_PAGE }, (_, i) => ({ + id: `q_${pad(i)}`, name: `queue_${pad(i)}`, label: `Queue ${i}`, organization_id: 'org_a', + })); + tables.sys_queue_member = [ + { id: 'qm_last', queue: `queue_${LAST}`, recipient_type: 'user', recipient_id: 'u_z', organization_id: 'org_a' }, + ]; + const svc = service(makeEngine({ tables })); + expect(await svc.queuesOf('u_z', 'org_a')).toEqual([`queue_${LAST}`]); + expect(await svc.hasAnyQueue('org_a')).toBe(true); + }); + + it('the queues naming a recipient are found past one page of member rows', async () => { + const tables = BASE_TABLES(); + tables.sys_queue_member = Array.from({ length: PAST_ONE_PAGE }, (_, i) => ({ + id: `qm_${pad(i)}`, queue: `queue_${pad(i)}`, recipient_type: 'position', + recipient_id: i === PAST_ONE_PAGE - 1 ? 'closer' : 'filler', organization_id: 'org_a', + })); + const svc = service(makeEngine({ tables })); + expect(await svc.queuesNaming('position', new Set(['closer']), 'org_a')).toEqual([`queue_${LAST}`]); + }); + + it('a read that fails on its second page answers as a failed read does today, never with the first page', async () => { + const engine = makeEngine({ tables: bigQueue() }); + const find = engine.find.bind(engine); + engine.find = async (object: string, options?: any) => { + if (object === 'sys_queue_member' && options?.where?.$and) throw new Error('sys_queue_member page 2 unavailable'); + return find(object, options); + }; + const warn = vi.fn(); + const svc = service(engine, { warn }); + + expect(await svc.expandMembers('big', 'org_a')).toEqual([]); + expect(await svc.isMember('u_00000', 'big', 'org_a')).toBe(false); + expect(warn).toHaveBeenCalledTimes(1); + expect(warn.mock.calls[0][0]).toMatch(/sys_queue_member read failed/); + expect(warn.mock.calls[0][1]).toEqual({ error: 'sys_queue_member page 2 unavailable' }); + }); + + it('a read that cannot vouch for completion logs its reason, not the walk\'s "refuses instead" — the queue answers nobody', async () => { + // A full page whose last row carries no id: the walk cannot resume after it. + const tables = bigQueue(10_000); + delete tables.sys_queue_member.at(-1)!.id; + const warn = vi.fn(); + const svc = service(makeEngine({ tables }), { warn }); + + expect(await svc.expandMembers('big', 'org_a')).toEqual([]); + expect(warn).toHaveBeenCalledTimes(1); + expect(warn.mock.calls[0][1].error).toBe( + "[sharing] the member rows of queue 'big' in organization 'org_a' could not be read: " + + 'the read by id could not continue past a page (a row without an id, or a page that repeated the cursor).', + ); + }); +}); + describe('[ADR-0131 D4] OwnershipQueueService.hasAnyQueue — the session payload\'s one boolean', () => { const ROW_ONLY = (): Record => ({ sys_queue: [{ id: 'q1', name: 'billing', label: 'Billing', organization_id: 'org_a' }],