Skip to content

Commit 7db9dbc

Browse files
committed
improvement(egress): keep org egress outages from disabling connectors or failing degradable work
- slack native formatInput: an outage resolving the bot token degrades the event (no token) instead of failing it - connector content and member sync: an outage retries at the first backoff rung without touching the auto-disable breaker - selector credential bundle: rethrow an outage instead of reporting the connection unavailable - internal LLM Vertex credential: an outage returns 503 with the fixed egress message - webhook provider subscriptions: projected failures keep the unavailable classification; the webhook save route and deploy report it as 503 - gmail/outlook configurePolling: rethrow an outage instead of the permissions rollback message
1 parent cfd06e4 commit 7db9dbc

20 files changed

Lines changed: 309 additions & 20 deletions

‎apps/sim/app/api/webhooks/route.test.ts‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,8 @@ vi.mock('@/lib/webhooks/utils.server', () => ({
5050
findConflictingWebhookPathOwner: mocks.findConflictingWebhookPathOwner,
5151
}))
5252

53+
import { OutboundRoutingError } from '@/lib/core/network/routing'
54+
import { EGRESS_UNAVAILABLE_MESSAGE, OrchestrationError } from '@/lib/core/orchestration/types'
5355
import { DEFAULT_PERMISSION_GROUP_CONFIG } from '@/lib/permission-groups/fields'
5456
import { POST } from '@/app/api/webhooks/route'
5557

@@ -619,3 +621,34 @@ describe('POST /api/webhooks subscription replacement recovery', () => {
619621
}
620622
)
621623
})
624+
625+
describe('POST /api/webhooks egress outage', () => {
626+
beforeEach(setupUpsertMocks)
627+
628+
it('reports an outage while creating the external subscription as unavailable', async () => {
629+
queueCreatePathRows()
630+
mocks.createExternalWebhookSubscription.mockRejectedValue(
631+
new OrchestrationError('unavailable', EGRESS_UNAVAILABLE_MESSAGE)
632+
)
633+
634+
const response = await POST(upsertRequest(), {})
635+
636+
expect(response.status).toBe(503)
637+
expect(await response.json()).toEqual({ error: EGRESS_UNAVAILABLE_MESSAGE })
638+
})
639+
640+
it('reports an outage while configuring polling as unavailable', async () => {
641+
queueCreatePathRows()
642+
dbChainMockFns.returning.mockImplementationOnce(async () => [
643+
{ id: 'webhook-1', workflowId: 'workflow-1', path: 'inbound-orders', isActive: true },
644+
])
645+
mocks.activateExternalWebhookSubscription.mockResolvedValue(undefined)
646+
mocks.configurePolling.mockRejectedValue(new OutboundRoutingError('GATEWAY_UNAVAILABLE'))
647+
mocks.getProviderHandler.mockReturnValue({ configurePolling: mocks.configurePolling })
648+
649+
const response = await POST(upsertRequest(), {})
650+
651+
expect(response.status).toBe(503)
652+
expect(await response.json()).toEqual({ error: EGRESS_UNAVAILABLE_MESSAGE })
653+
})
654+
})

‎apps/sim/app/api/webhooks/route.ts‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,11 @@ import { parseRequest } from '@/lib/api/server'
1717
import { getSession } from '@/lib/auth'
1818
import { authorizeCredentialUseForAuth } from '@/lib/auth/credential-access'
1919
import { AuthType } from '@/lib/auth/hybrid'
20+
import {
21+
EGRESS_UNAVAILABLE_MESSAGE,
22+
isUnavailableFailure,
23+
statusForOrchestrationError,
24+
} from '@/lib/core/orchestration/types'
2025
import { PlatformEvents } from '@/lib/core/telemetry'
2126
import { generateRequestId } from '@/lib/core/utils/request'
2227
import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
@@ -569,6 +574,13 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
569574
}
570575
}
571576
} catch (err) {
577+
if (isUnavailableFailure(err)) {
578+
logger.warn('External webhook subscription unavailable during an egress outage')
579+
return NextResponse.json(
580+
{ error: EGRESS_UNAVAILABLE_MESSAGE },
581+
{ status: statusForOrchestrationError('unavailable') }
582+
)
583+
}
572584
const isConfigurationError = err instanceof WebhookDeploymentConfigurationError
573585
if (isConfigurationError) {
574586
logger.warn('Rejected webhook trigger configuration', {
@@ -746,6 +758,12 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
746758
err
747759
)
748760
await revertSavedWebhook(savedWebhook, existingWebhook, requestId)
761+
if (isUnavailableFailure(err)) {
762+
return NextResponse.json(
763+
{ error: EGRESS_UNAVAILABLE_MESSAGE },
764+
{ status: statusForOrchestrationError('unavailable') }
765+
)
766+
}
749767
return NextResponse.json(
750768
{
751769
error: `Failed to configure ${provider} webhook`,

‎apps/sim/lib/internal/llm/operations.test.ts‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,8 @@ import { permissionCheckMock } from '@sim/testing/mocks/permission-check.mock'
1414
import { permissionsMock, permissionsMockFns } from '@sim/testing/mocks/permissions.mock'
1515
import { providersMock, providersMockFns } from '@sim/testing/mocks/providers.mock'
1616
import { beforeEach, describe, expect, it, vi } from 'vitest'
17+
import { OutboundRoutingError } from '@/lib/core/network/routing'
18+
import { EGRESS_UNAVAILABLE_MESSAGE } from '@/lib/core/orchestration/types'
1719

1820
const mocks = vi.hoisted(() => ({
1921
authorizeCredential: vi.fn(),
@@ -132,6 +134,26 @@ describe('executeLlmProviderOperation', () => {
132134
)
133135
})
134136

137+
it('reports a Vertex credential the egress outage kept from resolving as unavailable', async () => {
138+
mockExecuteProviderRequest.mockClear()
139+
mocks.resolveVertexAccessToken.mockRejectedValueOnce(
140+
new OutboundRoutingError('GATEWAY_UNAVAILABLE')
141+
)
142+
143+
await expect(
144+
executeLlmProviderOperation(
145+
{
146+
provider: 'vertex',
147+
model: 'vertex/gemini-2.5-pro',
148+
vertexCredential: 'credential-1',
149+
workspaceId: 'workspace-1',
150+
},
151+
{ actorUserId: 'user-1', headers: new Headers(), requestId: 'request-1' }
152+
)
153+
).rejects.toMatchObject({ status: 503, body: { error: EGRESS_UNAVAILABLE_MESSAGE } })
154+
expect(mockExecuteProviderRequest).not.toHaveBeenCalled()
155+
})
156+
135157
it("signs a Bedrock call with the selected AWS credential's session for the caller", async () => {
136158
mocks.authorizeCredential.mockResolvedValueOnce(AUTHORIZED_AWS_CREDENTIAL)
137159
mockResolveAuthorizedAwsCredential.mockResolvedValueOnce({

‎apps/sim/lib/internal/llm/operations.ts‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,11 @@ import {
88
type BillingAttributionSnapshot,
99
requireBillingAttributionHeader,
1010
} from '@/lib/billing/core/attribution'
11+
import {
12+
EGRESS_UNAVAILABLE_MESSAGE,
13+
isUnavailableFailure,
14+
statusForOrchestrationError,
15+
} from '@/lib/core/orchestration/types'
1116
import {
1217
inspectModelInputProjectionState,
1318
inspectModelInputProvenanceRequest,
@@ -158,6 +163,13 @@ async function prepareProviderRequest(
158163
if (error instanceof BedrockCredentialError && error.reason === 'unavailable') {
159164
fail(503, error.message)
160165
}
166+
if (isUnavailableFailure(error)) {
167+
logger.warn('Provider credential unavailable during an egress outage', {
168+
requestId: context.requestId,
169+
provider: input.provider,
170+
})
171+
fail(statusForOrchestrationError('unavailable'), EGRESS_UNAVAILABLE_MESSAGE)
172+
}
161173
logger.error('Failed to resolve provider credential', {
162174
requestId: context.requestId,
163175
provider: input.provider,

‎apps/sim/lib/knowledge/connectors/member-sync-engine.test.ts‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,8 @@ import { knowledgeDocumentsServiceMock } from '@sim/testing/mocks/knowledge-docu
44
import { knowledgeMemberAccessMock } from '@sim/testing/mocks/knowledge-member-access.mock'
55
import { DrizzleQueryError } from 'drizzle-orm/errors'
66
import { beforeEach, describe, expect, it, vi } from 'vitest'
7+
import { OutboundRoutingError } from '@/lib/core/network/routing'
8+
import { EGRESS_UNAVAILABLE_MESSAGE } from '@/lib/core/orchestration/types'
79

810
vi.mock('@/connectors/registry.server', () => ({ CONNECTOR_REGISTRY: {} }))
911
vi.mock('@/lib/knowledge/documents/service', () => knowledgeDocumentsServiceMock)
@@ -273,6 +275,19 @@ describe('member sync engine decisions', () => {
273275
})
274276
expect(update.nextMemberSyncAt).not.toBeNull()
275277
})
278+
279+
it('does not disable a connector one failure from the breaker over an egress outage', async () => {
280+
const update = await resolveMemberSyncFailureUpdate(
281+
new Error('listing failed', { cause: new OutboundRoutingError('GATEWAY_UNAVAILABLE') }),
282+
failure
283+
)
284+
expect(update).toMatchObject({
285+
memberSyncStatus: 'error',
286+
lastMemberSyncError: EGRESS_UNAVAILABLE_MESSAGE,
287+
memberSyncConsecutiveFailures: MAX_CONSECUTIVE_FAILURES - 1,
288+
})
289+
expect(update.nextMemberSyncAt).not.toBeNull()
290+
})
276291
})
277292

278293
describe('nextMemberSyncTime', () => {

‎apps/sim/lib/knowledge/connectors/member-sync-engine.ts‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ import {
2828
type BillingAttributionSnapshot,
2929
} from '@/lib/billing/core/attribution'
3030
import { withResourceOutboundScope } from '@/lib/core/network/resource-scope.server'
31+
import { EGRESS_UNAVAILABLE_MESSAGE, isUnavailableFailure } from '@/lib/core/orchestration/types'
3132
import { resourceScopeFields, resourceScopeFromOwner } from '@/lib/core/resource-scope'
3233
import { resourceScopeCondition } from '@/lib/core/resource-scope.server'
3334
import {
@@ -382,8 +383,8 @@ export function memberRunMadeProgress(result: MemberSyncResult): boolean {
382383

383384
/**
384385
* The connector row a failed members-mode run writes. A deterministic capacity rejection waits for
385-
* an operator, a transient database failure retries without touching the breaker, and anything
386-
* else climbs the ladder toward auto-disable.
386+
* an operator, a transient database failure or an organization egress outage retries without
387+
* touching the breaker, and anything else climbs the ladder toward auto-disable.
387388
*/
388389
export async function resolveMemberSyncFailureUpdate(
389390
error: unknown,
@@ -423,6 +424,14 @@ export async function resolveMemberSyncFailureUpdate(
423424
})
424425
)
425426
}
427+
if (isUnavailableFailure(error)) {
428+
return buildMemberSyncDatabaseRetryUpdate(
429+
now,
430+
failure.previousFailures,
431+
EGRESS_UNAVAILABLE_MESSAGE,
432+
connectorFailureBackoffMinutes(1) * 60 * 1000
433+
)
434+
}
426435
return buildMemberSyncFailureUpdate(
427436
now,
428437
failure.previousFailures,

‎apps/sim/lib/knowledge/connectors/sync-engine.test.ts‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@ import {
2525
import { generateShortId } from '@sim/utils/id'
2626
import { DrizzleQueryError } from 'drizzle-orm/errors'
2727
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
28+
import { OutboundRoutingError } from '@/lib/core/network/routing'
29+
import { EGRESS_UNAVAILABLE_MESSAGE } from '@/lib/core/orchestration/types'
2830
import * as connectorTokens from '@/lib/knowledge/connectors/access-token'
2931
import {
3032
routeWindowTallies,
@@ -820,6 +822,19 @@ describe('executeSync database failures', () => {
820822
expect.objectContaining({ status: 'failed', databaseFailureClass: 'capacity' })
821823
)
822824
})
825+
826+
it('does not disable a connector one failure from the breaker over an egress outage', async () => {
827+
const { terminal, MAX_CONSECUTIVE_FAILURES } = await failSyncWith(
828+
new OutboundRoutingError('GATEWAY_UNAVAILABLE')
829+
)
830+
831+
expect(terminal).toMatchObject({
832+
status: 'error',
833+
lastSyncError: EGRESS_UNAVAILABLE_MESSAGE,
834+
consecutiveFailures: MAX_CONSECUTIVE_FAILURES - 1,
835+
})
836+
expect((terminal?.nextSyncAt as Date).getTime()).toBeGreaterThan(NOW.getTime())
837+
})
823838
})
824839

825840
describe('previous complete listing evidence', () => {

‎apps/sim/lib/knowledge/connectors/sync-engine.ts‎

Lines changed: 18 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import {
2020
type BillingAttributionSnapshot,
2121
} from '@/lib/billing/core/attribution'
2222
import { withResourceOutboundScope } from '@/lib/core/network/resource-scope.server'
23+
import { EGRESS_UNAVAILABLE_MESSAGE, isUnavailableFailure } from '@/lib/core/orchestration/types'
2324
import {
2425
type ResourceScope,
2526
resourceScopeFields,
@@ -1645,19 +1646,26 @@ export async function executeSync(
16451646
madeProgress: result.docsAdded + result.docsUpdated + result.docsDeleted > 0,
16461647
})
16471648
)
1648-
: rateLimited
1649-
? buildSyncRateLimitUpdate(
1649+
: isUnavailableFailure(error)
1650+
? buildSyncDatabaseRetryUpdate(
16501651
new Date(),
16511652
connector.consecutiveFailures,
1652-
errorMessage,
1653-
retryAfterMs
1654-
)
1655-
: buildSyncFailureUpdate(
1656-
new Date(),
1657-
connector.consecutiveFailures,
1658-
errorMessage,
1659-
retryAfterMs
1653+
EGRESS_UNAVAILABLE_MESSAGE,
1654+
connectorFailureBackoffMinutes(1) * 60 * 1000
16601655
)
1656+
: rateLimited
1657+
? buildSyncRateLimitUpdate(
1658+
new Date(),
1659+
connector.consecutiveFailures,
1660+
errorMessage,
1661+
retryAfterMs
1662+
)
1663+
: buildSyncFailureUpdate(
1664+
new Date(),
1665+
connector.consecutiveFailures,
1666+
errorMessage,
1667+
retryAfterMs
1668+
)
16611669

16621670
if (failureUpdate.status === 'disabled') {
16631671
logger.warn('Connector disabled after repeated failures', {

‎apps/sim/lib/selectors/server/providers/credential-bundle.test.ts‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import { authOAuthUtilsMock, authOAuthUtilsMockFns } from '@sim/testing/mocks/auth-oauth-utils.mock'
22
import { describe, expect, it, vi } from 'vitest'
3+
import { OutboundRoutingError } from '@/lib/core/network/routing'
34

45
const mockResolveOrganizationToken = vi.hoisted(() => vi.fn())
56
vi.mock('@/lib/credentials/application/organization-credentials', () => ({
@@ -36,6 +37,21 @@ describe('selector credential bundles', () => {
3637
expect(protectedValues.contains('prefix-cloud-1-suffix')).toBe(false)
3738
})
3839

40+
it('surfaces an egress outage instead of reporting the connection unavailable', async () => {
41+
const outage = new OutboundRoutingError('GATEWAY_UNAVAILABLE')
42+
mockResolveCredentialAccessToken.mockRejectedValueOnce(outage)
43+
44+
await expect(
45+
resolveSelectorCredentialBundle({
46+
credential: {
47+
suppliedId: 'credential-1',
48+
access: { ok: true, credentialOwnerUserId: 'owner-1' },
49+
},
50+
protectedValues: createSelectorProtectedValues(),
51+
})
52+
).rejects.toBe(outage)
53+
})
54+
3955
it('preserves a selector abort without canceling the shared resolution', async () => {
4056
let resolveShared!: (value: { accessToken: string }) => void
4157
const sharedResolution = new Promise<{ accessToken: string }>((resolve) => {

‎apps/sim/lib/selectors/server/providers/credential-bundle.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
import { isUnavailableFailure } from '@/lib/core/orchestration/types'
12
import { resolveOrganizationCredentialTokenBundle } from '@/lib/credentials/application/organization-credentials'
23
import {
34
resolveCredentialTokenBundle,
@@ -61,7 +62,7 @@ export async function resolveSelectorCredentialBundle(input: {
6162
)
6263
credential.signal?.throwIfAborted()
6364
} catch (error) {
64-
if (credential.signal?.aborted) throw error
65+
if (credential.signal?.aborted || isUnavailableFailure(error)) throw error
6566
throw new SelectorConnectionUnavailableError()
6667
}
6768
if (!bundle?.accessToken) throw new SelectorConnectionUnavailableError()

0 commit comments

Comments
 (0)