fix(service-automation): a connection already marked reconnect required answers the cause it stores, and the refusal log names it

.changeset/28963-reconnect-required-stored-cause.md

@@ -0,0 +1,14 @@
+---
+'@objectstack/service-automation': patch
+---
+
+fix(service-automation): a connection already marked "reconnect required" answers the cause it stores, and the refusal log names it (#28963)
+
+Clause-②: no — the text of a dispatch refusal on an already-marked connection and of one log line. No accept set, status value, error code or export changes meaning.
+Metadata-protocol: none
+
+A refresh that marks a connector connection `reconnect_required` stores a reason naming its cause: the authorization server refused the grant, it issued no refresh token, or the connection keeps no client to refresh with. Two places still said the server refused the grant whatever had happened.
+
+- **Every later dispatch on the marked connection** answers the reason the row stores, the same sentence the first dispatch answered. Before, only the dispatch that marked the row named the cause; every one after it (another replica, a restart, a retry) said the authorization server refused the grant. A row that stores no reason still answers the refusal sentence.
+- **The server log line** written when a refresh marks a connection is now that stored reason, so it names the cause on each of the three refresh paths. Before, it said "the authorization server refused the grant" on all three.
+- Unchanged: the 503 `SERVICE_UNAVAILABLE` refusal itself, the reason each path stores, the instance's "reconnect required" hold, and every status, code and write.

packages/services/service-automation/src/connector-connection-door.test.ts

@@ -62,6 +62,8 @@ const idp = {
     requests: [] as TokenRequest[],
     /** Answer to a refresh_token grant; default rotate. */
     refresh: 'rotate' as 'rotate' | 'invalid_grant' | 'down' | 'no_rotate',
+    /** Whether the code exchange issues a refresh token; default yes. */
+    exchangeIssuesRefreshToken: true,
     issued: 0,
 };
 
@@ -86,7 +88,11 @@ beforeAll(async () => {
         if (form.grant_type === 'authorization_code') {
             if (form.code !== GOOD_CODE) return answer(400, { error: 'invalid_grant' });
             return answer(200, {
-                access_token: `dummy-at-exchange-${n}`, token_type: 'Bearer', expires_in: 3600, refresh_token: `dummy-rt-${n}`, scope: 'crm.read',
+                access_token: `dummy-at-exchange-${n}`,
+                token_type: 'Bearer',
+                expires_in: 3600,
+                ...(idp.exchangeIssuesRefreshToken ? { refresh_token: `dummy-rt-${n}` } : {}),
+                scope: 'crm.read',
             });
         }
         if (form.grant_type === 'refresh_token') {
@@ -200,6 +206,7 @@ afterEach(async () => {
     }
     idp.requests.length = 0;
     idp.refresh = 'rotate';
+    idp.exchangeIssuesRefreshToken = true;
     idp.issued = 0;
     vi.restoreAllMocks();
 });
@@ -490,6 +497,25 @@ describe('ADR-0143 D4 — dispatch with the connection', () => {
         expect((await rowsOf(ql))[0]).toMatchObject({ status: 'connected', revision: 2 });
     });
 
+    it('[#28963] a connection marked because no refresh token was issued: the log and the hold name that cause, not a refusal', async () => {
+        const out = captureOutput();
+        const { call, connect, provider, engine, ql } = await boot();
+        idp.exchangeIssuesRefreshToken = false;
+        await (await connect()).done();
+        provider.state.accepted = new Set();
+        const refused = await call('org_a').catch((e) => e);
+        expect(refused).toMatchObject({ status: 503, message: expect.stringContaining('issued no refresh token') });
+        const [row] = await rowsOf(ql);
+        expect(row).toMatchObject({ status: 'reconnect_required', status_reason: refused.message });
+        // The log line is the stored reason, whose cause is the one that happened.
+        await vi.waitFor(() => expect(out.text()).toContain(`[Automation] ${refused.message}.`));
+        expect(out.text()).not.toContain('refused the grant');
+        // The instance is held with the cause the row stores.
+        await vi.waitFor(() => expect(engine.getConnectorDegradedReason('crm')).toContain('issued no refresh token'));
+        expect(engine.getConnectorDegradedReason('crm')).not.toContain('refused the grant');
+        expect(idp.requests.filter((r) => r.form.grant_type === 'refresh_token')).toHaveLength(0);
+    });
+
     it("the run's organization selects the connection; a run with none is refused when several exist", async () => {
         const { call, connect, provider } = await boot();
         await (await connect(ADMIN_A)).done();
@@ -573,7 +599,9 @@ describe('ADR-0143 D3/D6 — what a grant leaves behind', () => {
         }
         // The logs did record the connect — the pin is not passing on silence.
         expect(out.text()).toContain("connector instance 'crm' connected for organization 'org_a' by user 'usr_admin'");
-        expect(out.text()).toContain("refused the grant of organization 'org_a'");
+        expect(out.text()).toContain(
+            "[Automation] connector instance 'crm' needs to be reconnected for organization 'org_a': its authorization server refused the grant",
+        );
     });
 
     it('⛔ writes only its own record — never sys_metadata, never a sign-in account (D6) — and reads no account either', async () => {

packages/services/service-automation/src/oauth2-grant/grant-handle.test.ts

@@ -11,7 +11,7 @@ import { callWithBorrowedCredential, BORROWED_CREDENTIAL_SKEW_MS } from '@object
 import { ConnectorUpstreamUnavailableError } from '@objectstack/spec/integration';
 import type { ILock } from '@objectstack/spec/contracts';
 import type { ConnectorConnectionRecord, ConnectorConnectionStore, ConnectorGrantWrite } from '../connector-connections.js';
-import { createOAuth2GrantHandle, type OAuth2GrantConfig } from './grant-handle.js';
+import { createOAuth2GrantHandle, reconnectRequiredReason, type OAuth2GrantConfig } from './grant-handle.js';
 
 const TOKEN_URL = 'https://idp.example.com/oauth/token';
 
@@ -308,7 +308,7 @@ describe('ADR-0143 D4 — authorization_code: refresh', () => {
         expect(refused.message).not.toContain('dummy-rt-a');
         expect(calls.markReconnectRequired).toEqual(['org_a']);
         expect(held.get('org_a')).toMatchObject({ status: 'reconnect_required', refreshToken: undefined });
-        expect(onGrantRefused).toHaveBeenCalledWith('org_a');
+        expect(onGrantRefused).toHaveBeenCalledWith('org_a', refused.message);
         // The next dispatch reads the row: refused before any token request.
         await expect(handle.get({ organizationId: 'org_a' })).rejects.toMatchObject({ status: 503 });
         expect(idp.state.requests).toHaveLength(1);
@@ -410,7 +410,7 @@ describe('ADR-0143 D2.1 — an MCP instance whose connection keeps the client it
         await expect(handle.get({ organizationId: 'org_a' })).rejects.toMatchObject({ code: 'SERVICE_UNAVAILABLE', status: 503 });
         expect(idp.seen).toHaveLength(0);
         expect(calls.markReconnectRequired).toEqual(['org_a']);
-        expect(onGrantRefused).toHaveBeenCalledWith('org_a');
+        expect(onGrantRefused).toHaveBeenCalledWith('org_a', expect.stringContaining('keeps no client to refresh with'));
     });
 });
 
@@ -458,3 +458,94 @@ describe('[#28863] the stored reason of a connection marked reconnect required n
         expect(idp.state.requests).toHaveLength(1);
     });
 });
+
+describe('[#28963] a connection already marked reconnect required answers the cause it stores, and the log names it', () => {
+    const NO_REFRESH_TOKEN = 'issued no refresh token, so an access token the connection has to renew cannot be renewed';
+    const NO_CLIENT = 'keeps no client to refresh with, so an access token it has to renew cannot be renewed';
+    const REFUSED = 'its authorization server refused the grant (revoked, or expired with no usable refresh)';
+    const refusalOf = (p: Promise<unknown>) => p.then(() => { throw new Error('the dispatch was not refused'); }, (e: Error & { status?: number; code?: string }) => e);
+
+    it('a row marked because no refresh token was issued answers that sentence on the first dispatch and on every later one', async () => {
+        const idp = tokenEndpoint();
+        const { store, held } = fakeConnections([{ org: 'org_a' }]);
+        const handle = createOAuth2GrantHandle({ connector: 'crm', config: AUTH_CODE, connections: store, fetch: idp.fetch });
+        const first = await refusalOf(handle.get({ organizationId: 'org_a' }));
+        const stored = String(held.get('org_a')?.statusReason);
+        expect(stored).toContain(NO_REFRESH_TOKEN);
+        expect(first.message).toBe(stored);
+        // Every later dispatch reads the marked row and answers what it stores, not the refusal sentence.
+        for (const later of [
+            () => handle.get({ organizationId: 'org_a' }),
+            () => handle.get({ organizationId: 'org_a' }),
+            () => handle.refresh({ organizationId: 'org_a' }),
+        ]) {
+            const refused = await refusalOf(later());
+            expect(refused).toMatchObject({ status: 503, code: 'SERVICE_UNAVAILABLE' });
+            expect(refused.message).toBe(stored);
+            expect(refused.message).not.toMatch(/refused/);
+        }
+        // So does a handle that never saw the marking (another replica, a restart).
+        const other = createOAuth2GrantHandle({ connector: 'crm', config: AUTH_CODE, connections: store, fetch: idp.fetch });
+        expect((await refusalOf(other.get({ organizationId: 'org_a' }))).message).toBe(stored);
+        expect(idp.state.requests).toHaveLength(0);
+    });
+
+    it('a row another replica marked while this one waited for the lock answers the cause that replica stored', async () => {
+        const idp = tokenEndpoint();
+        const { store, held } = fakeConnections([{ org: 'org_a', refreshToken: 'dummy-rt-a' }]);
+        const markedThere = reconnectRequiredReason('crm', 'org_a', 'no_client');
+        const lock = {
+            async withLock(_key: string, fn: () => Promise<unknown>) {
+                Object.assign(held.get('org_a')!, { status: 'reconnect_required', statusReason: markedThere, refreshToken: undefined });
+                return fn();
+            },
+        } as unknown as ILock;
+        const handle = createOAuth2GrantHandle({ connector: 'crm', config: AUTH_CODE, connections: store, lock: () => lock, fetch: idp.fetch });
+        const refused = await refusalOf(handle.get({ organizationId: 'org_a' }));
+        expect(refused).toMatchObject({ status: 503, code: 'SERVICE_UNAVAILABLE' });
+        expect(refused.message).toBe(markedThere);
+        expect(idp.state.requests).toHaveLength(0);
+    });
+
+    it('a marked row that stores no cause still answers the refusal sentence, on both reads', async () => {
+        const idp = tokenEndpoint();
+        const { store, held } = fakeConnections([{ org: 'org_a', status: 'reconnect_required' }]);
+        const handle = createOAuth2GrantHandle({ connector: 'crm', config: AUTH_CODE, connections: store, fetch: idp.fetch });
+        const atSelect = await refusalOf(handle.get({ organizationId: 'org_a' }));
+        expect(atSelect.message).toBe(reconnectRequiredReason('crm', 'org_a'));
+        expect(atSelect.message).toContain(REFUSED);
+
+        held.get('org_a')!.status = 'connected';
+        held.get('org_a')!.refreshToken = 'dummy-rt-a';
+        const lock = {
+            async withLock(_key: string, fn: () => Promise<unknown>) {
+                held.get('org_a')!.status = 'reconnect_required';
+                return fn();
+            },
+        } as unknown as ILock;
+        const locked = createOAuth2GrantHandle({ connector: 'crm', config: AUTH_CODE, connections: store, lock: () => lock, fetch: idp.fetch });
+        const underLock = await refusalOf(locked.get({ organizationId: 'org_a' }));
+        expect(underLock.message).toBe(reconnectRequiredReason('crm', 'org_a'));
+        expect(idp.state.requests).toHaveLength(0);
+    });
+
+    it('the plugin is told the cause of each refresh path, in the words it stored', async () => {
+        const cases = [
+            { cause: NO_REFRESH_TOKEN, config: AUTH_CODE, row: { org: 'org_a' }, answer: undefined },
+            { cause: NO_CLIENT, config: { grant: 'authorization_code' } as OAuth2GrantConfig, row: { org: 'org_a', refreshToken: 'dummy-rt-a' }, answer: undefined },
+            { cause: REFUSED, config: AUTH_CODE, row: { org: 'org_a', refreshToken: 'dummy-rt-a' }, answer: { status: 400, body: { error: 'invalid_grant' } } },
+        ];
+        for (const { cause, config, row, answer } of cases) {
+            const idp = tokenEndpoint();
+            if (answer) idp.state.answer = () => answer;
+            const { store, held } = fakeConnections([row]);
+            const onGrantRefused = vi.fn();
+            const handle = createOAuth2GrantHandle({ connector: 'crm', config, connections: store, fetch: idp.fetch, onGrantRefused });
+            await refusalOf(handle.get({ organizationId: 'org_a' }));
+            const stored = String(held.get('org_a')?.statusReason);
+            expect(stored).toContain(cause);
+            expect(onGrantRefused).toHaveBeenCalledTimes(1);
+            expect(onGrantRefused).toHaveBeenCalledWith('org_a', stored);
+        }
+    });
+});

packages/services/service-automation/src/oauth2-grant/grant-handle.ts

@@ -78,11 +78,12 @@ export interface OAuth2GrantHandleDeps {
     /** The cluster lock, when the deployment has a cluster service. Read at refresh time. */
     readonly lock?: () => ILock | undefined;
     /**
-     * Told when the authorization server refused an organization's grant and
-     * the connection was marked `reconnect_required`, so the plugin re-judges
+     * Told when a refresh marked an organization's connection
+     * `reconnect_required`, with the `reason` it stored (whose cause is the
+     * one that happened on that path), so the plugin logs it and re-judges
      * the instance. A throw here is the listener's own.
      */
-    readonly onGrantRefused?: (organizationId: string) => void;
+    readonly onGrantRefused?: (organizationId: string, reason: string) => void;
     readonly timeoutMs?: number;
     readonly fetch?: typeof fetch;
     readonly now?: () => number;
@@ -155,6 +156,15 @@ export function reconnectRequiredReason(
         + 'uses the new connection (ADR-0143 D4)';
 }
 
+/**
+ * The dispatch refusal of a connection already marked `reconnect_required`:
+ * the reason the row stores, whose cause is the one that marked it, and the
+ * authorization server's refusal only for a row that stores none.
+ */
+function markedReason(connector: string, record: ConnectorConnectionRecord): string {
+    return record.statusReason ?? reconnectRequiredReason(connector, record.organizationId);
+}
+
 /** Create the handle of one `oauth2` instance. */
 export function createOAuth2GrantHandle(deps: OAuth2GrantHandleDeps): OAuth2GrantHandle {
     const { connector, config } = deps;
@@ -235,7 +245,7 @@ export function createOAuth2GrantHandle(deps: OAuth2GrantHandleDeps): OAuth2Gran
             throw unavailable(`the connection of connector instance '${connector}' could not be read now; the next dispatch retries`);
         }
         if (!record) throw unavailable(notConnectedReason(connector, organizationId));
-        if (record.status === 'reconnect_required') throw unavailable(reconnectRequiredReason(connector, record.organizationId));
+        if (record.status === 'reconnect_required') throw unavailable(markedReason(connector, record));
         return record;
     };
 
@@ -248,7 +258,7 @@ export function createOAuth2GrantHandle(deps: OAuth2GrantHandleDeps): OAuth2Gran
             /* the refusal stands either way; the row is re-judged on the next dispatch */
         }
         try {
-            deps.onGrantRefused?.(record.organizationId);
+            deps.onGrantRefused?.(record.organizationId, reason);
         } catch {
             /* the listener's throw is its own */
         }
@@ -286,7 +296,7 @@ export function createOAuth2GrantHandle(deps: OAuth2GrantHandleDeps): OAuth2Gran
     const refreshOnce = async (organizationId: string): Promise<CachedToken> => {
         const current = await deps.connections.read(organizationId, connector);
         if (!current) throw unavailable(notConnectedReason(connector, organizationId));
-        if (current.status === 'reconnect_required') throw unavailable(reconnectRequiredReason(connector, organizationId));
+        if (current.status === 'reconnect_required') throw unavailable(markedReason(connector, current));
         let refreshToken: string | undefined;
         try {
             refreshToken = await deps.connections.resolveRefreshToken(current);

packages/services/service-automation/src/plugin.ts

@@ -3692,11 +3692,10 @@ export class AutomationServicePlugin implements Plugin {
             },
             connections: this.connectorConnectionStoreOrNone(),
             lock: () => this.clusterService(ctx)?.lock,
-            onGrantRefused: (organizationId) => {
-                ctx.logger.warn(
-                    `[Automation] connector instance '${name}': the authorization server refused the grant of organization `
-                        + `'${organizationId}'; it needs to be reconnected there (ADR-0143 D4).`,
-                );
+            onGrantRefused: (_organizationId, reason) => {
+                // The reason the connection stored: it names the instance, the
+                // organization and the cause that happened on that refresh path.
+                ctx.logger.warn(`[Automation] ${reason}.`);
                 void this.materializeDeclaredConnectors(ctx, { fatal: false }).catch(() => { /* logged by the reconcile */ });
             },
             ...(typeof entry.requestTimeoutMs === 'number' ? { timeoutMs: entry.requestTimeoutMs } : {}),