feat: integration builder POC [CM-1372] - #4463
Conversation
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
PR SummaryMedium Risk Overview A 30s dispatcher workflow claims due units (lease + The new Ops wiring: Docker compose service, Dockerfile, and Reviewed by Cursor Bugbot for commit 032d4cd. Bugbot is set up for automated code reviews on this repo. Configure here. |
|
|
mbani01
changed the title
There was a problem hiding this comment.
Pull request overview
Adds a proof-of-concept persistence and scheduling layer for integration sync units.
Changes:
- Defines sync-unit types and statuses.
- Adds claiming, rescheduling, and run-recording queries.
- Creates the sync-unit table and due-work index.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
services/libs/data-access-layer/src/integrationBuilder/types.ts |
Defines sync-unit data contracts. |
services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts |
Implements sync-unit database operations. |
backend/src/database/migrations/V1786442761__createSyncUnitsTable.sql |
Adds sync-unit storage and indexing. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
![github-license-compliance[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 8 out of 9 changed files in this pull request and generated no new comments.
Files not reviewed (1)
- pnpm-lock.yaml: Generated file
Suppressed comments (3)
services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts:48
- Soft-deleting an integration does not deactivate these rows:
IntegrationRepository.destroyuses the paranoid integration model, while this claim only checks the sync-unit status. As a result, disconnected integrations remain claimable and continue syncing indefinitely. Filter candidates to integrations whosedeletedAtis null (and separately decommission their units if retention requires it).
WHERE status = 'active'
AND "nextRunAt" <= now()
AND ("lockedAt" IS NULL OR "lockedAt" < now() - $(leaseMinutes) * interval '1 minute')
services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts:42
- This lease can expire and be claimed by a second worker, but the returned
lockedAtis not used as an ownership token byrescheduleUnit,recordRunSuccess, orrecordRunFailure; all three update byidalone. A slow first worker can therefore overwrite the newer run's watermark/counters and even clear its lock. Pass the claimed lease value (or a generated claim token) to every completion update, include it in theWHEREclause, and reject a zero-row update.
SET "lockedAt" = now(), "updatedAt" = now()
services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts:9
- These new data-access functions are not exported from the package entry point:
@crowd/data-access-layerresolves tosrc/index.ts, which has nointegrationBuilderexport, and this directory has no index module. Consumers therefore cannot use the normal package API and must rely on an internal/src/...deep import. Add anintegrationBuilder/index.tsbarrel and export it from the root index.
export async function upsertSyncUnits(qx: QueryExecutor, units: SyncUnitUpsert[]): Promise<number> {
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 8 out of 9 changed files in this pull request and generated 1 comment.
Files not reviewed (1)
- pnpm-lock.yaml: Generated file
Suppressed comments (2)
services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts:26
- This only inserts/renames discovered units. A channel or sync omitted by a later discovery remains
activeand continues being scheduled, while a previouslydecommissionedunit that is rediscovered remains decommissioned. Reconcile the complete discovered set transactionally: decommission missing units and reactivate rediscovered ones (while preserving intentionally paused/dead-letter units).
ON CONFLICT ("integrationId", "channelId", "syncName")
DO UPDATE SET "channelName" = EXCLUDED."channelName", "updatedAt" = now()`,
services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts:53
- Filtering soft-deleted integrations only at claim time leaves their units
activewith permanently overduenextRunAtvalues. Those rows stay at the front ofix_sync_units_due, so every scheduler poll must scan past an ever-growing set of unclaimable units. Decommission sync units as part of integration deletion (or add equivalent cleanup) so they leave the partial due index.
AND EXISTS (
SELECT 1
FROM public.integrations i
WHERE i.id = su2."integrationId" AND i."deletedAt" IS NULL
)
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 9 out of 10 changed files in this pull request and generated 1 comment.
Files not reviewed (1)
- pnpm-lock.yaml: Generated file
Suppressed comments (5)
services/libs/integration-builder/src/credentials.ts:31
- These lines document a future implementation change rather than a required invariant. Remove them; the credential-loading function and environment variable names are self-explanatory.
// POC only: secrets come from env; secret-manager adoption (OCI Vault) swaps this
// body without touching callers — getCredential stays the single entry point
services/libs/integration-builder/package.json:19
zodis not used anywhere in the new package, but adding it also introduces a separate Zod 3 installation in the lockfile. Remove the dependency until validation is implemented.
"zod": "^3.22.0"
services/libs/integration-builder/src/types.ts:14
- These POC/future-design notes describe the current change rather than a non-obvious invariant. Remove them; the fixed
kindtype already makes the temporary single-variant constraint clear.
// POC only: single variant; becomes a discriminated union (token, oauth2, ...)
// as more connectors land
services/libs/integration-builder/src/credentials.ts:20
- This change note only restates the temporary switch design and does not document an invariant. Remove it rather than retaining POC commentary in the implementation.
This issue also appears on line 30 of the same file.
// POC scope: GitHub only; each migrated connector adds its platform case here
services/libs/integration-builder/package.json:16
@crowd/commonis not imported anywhere in this new package. Remove the unused dependency so the package declares only its actual runtime requirements.
This issue also appears on line 19 of the same file.
"@crowd/common": "workspace:*",
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 11 out of 12 changed files in this pull request and generated no new comments.
Files not reviewed (1)
- pnpm-lock.yaml: Generated file
Suppressed comments (5)
services/libs/integration-builder/src/credentials.ts:12
- This lookup discards the integration-specific identity and returns the same global app credential for every GitHub integration. GitHub integrations are scoped by an installation ID (
integrationIdentifier), whilegithub-nangointegrations use mapped connection IDs; becauseManifest.discoverreceives only this credential, it cannot restrict discovery to the requested integration and can associate another installation's channels with it. Include the relevant installation/connection identity in the credential (or pass the integration identity into discovery) and handle the two platform credential models separately.
const integration: { platform: string } | null = await qx.selectOneOrNone(
`SELECT platform
FROM integrations
WHERE id = $(integrationId) AND "deletedAt" IS NULL`,
services/libs/integration-builder/src/credentials.ts:31
- This comment describes the implementation and an unticketed future replacement rather than an allowed invariant or external quirk. Remove it; the helper name and environment-variable reads are self-explanatory.
// POC only: secrets come from env; secret-manager adoption (OCI Vault) swaps this
// body without touching callers — getCredential stays the single entry point
services/libs/integration-builder/src/types.ts:15
- These POC/future-design notes do not document an external quirk, invariant, constraint, legacy complexity, or ticketed TODO. Remove them and let the credential type express the currently supported variant.
// POC only: single variant; becomes a discriminated union (token, oauth2, ...)
// as more connectors land
services/libs/integration-builder/src/credentials.ts:21
- This scope note describes the current implementation and future work without a ticket, which is not an allowed code-comment case. Remove it; the switch already makes the supported platforms clear.
This issue also appears on line 30 of the same file.
// POC scope: GitHub only; each migrated connector adds its platform case here
services/libs/data-access-layer/src/integrationBuilder/syncUnits.ts:40
- The new atomic-claim behavior has no database integration test. Add coverage that runs concurrent claims and verifies an ID is returned once, while active/due, deleted-integration, and expired-lease filtering behave as intended; comparable data-access SQL is exercised in
services/libs/data-access-layer/src/packages/*.integration.test.ts.
export async function claimDueUnits(qx: QueryExecutor, limit: number): Promise<ISyncUnit[]> {
return qx.select(
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 26 out of 27 changed files in this pull request and generated no new comments.
Files not reviewed (1)
- pnpm-lock.yaml: Generated file
Suppressed comments (7)
services/libs/connectors/src/credentials.ts:28
- This comment documents a planned secret-manager migration rather than an invariant callers must understand. Remove it or replace it with a ticketed
TODO(CM-XXX)when that migration is scheduled.
// POC only: secrets come from env; secret-manager adoption (OCI Vault) swaps this
// body without touching callers — getCredential stays the single entry point
services/apps/connectors_worker/src/activities/syncRunActivities.ts:65
- This roadmap comment explains a future error taxonomy rather than a non-obvious invariant. Remove it or use a ticketed
TODO(CM-XXX)for the M2 follow-up.
// POC only: everything unclassified is framework.internal; the 7-class
// error taxonomy arrives with the M2 HTTP client
services/libs/connectors/src/types.ts:14
- This roadmap note describes a future type change rather than a non-obvious invariant. Remove it; if the union work must be tracked, use a
TODO(CM-XXX)tied to a ticket.
// POC only: single variant; becomes a discriminated union (token, oauth2, ...)
// as more connectors land
services/libs/connectors/src/credentials.ts:17
- This POC-scope comment only restates the two switch cases and documents future work. Remove it; deferred connector work should be tracked with a ticket rather than an untracked source comment.
This issue also appears on line 27 of the same file.
// POC scope: GitHub only; each migrated connector adds its platform case here
services/libs/connectors/src/credentials.ts:10
- This raw integration lookup places database access in the connectors library, bypassing the repository's data-access-layer boundary. Move the query into
services/libs/data-access-layer/src/connectorsas a reusable function and call that exported function here; the fix spans both packages.
const integration: { platform: string } | null = await qx.selectOneOrNone(
`SELECT platform
FROM integrations
WHERE id = $(integrationId) AND "deletedAt" IS NULL`,
{ integrationId },
services/apps/connectors_worker/src/main.ts:33
- This comment is a roadmap note for the M4 connector rather than a required invariant. Remove it and track the future registration work in the referenced ticket.
// POC only: dummy connector drives the control-plane end-to-end; real
// connectors register here starting with GitHub in M4
services/apps/connectors_worker/src/activities/syncRunActivities.ts:38
- This comment only describes the current POC implementation and a planned sink. Remove it; future Kafka sink work should be represented by a ticketed
TODO(CM-XXX)if it must remain visible in code.
This issue also appears on line 64 of the same file.
// POC only: emit collects counts in memory; the Kafka sink emitter arrives in M2
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
There are 2 total unresolved issues (including 1 from previous review).
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit 9de976f. Configure here.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 29 out of 30 changed files in this pull request and generated 1 comment.
Files not reviewed (1)
- pnpm-lock.yaml: Generated file
Suppressed comments (11)
services/libs/data-access-layer/src/connectors/syncUnits.ts:88
- A successful run resets the failure count but leaves
lastErrorClasspopulated, so consumers will continue reporting a stale failure after recovery. Clear it together withconsecutiveFailures.
"consecutiveFailures" = 0,
"updatedAt" = now()
services/libs/connectors/src/http/client.ts:161
Retry-Afteris also valid on 503 responses. Treating the header alone as a rate limit parks a healthy credential and bypasses theprovider.unavailableretry path. Keep rate-limit detection status-specific here and useinterpretResponsefor provider-specific cases.
return 'retry-after' in headers
services/libs/connectors/src/http/client.ts:168
- HTTP permits
Retry-Afterto be an HTTP-date as well as delta-seconds. Date-form values currently fall through to the 60-second fallback, potentially retrying before the provider's requested time. Parse a valid future date before applying the fallback.
const retryAfterSeconds = Number(headers['retry-after'])
if (Number.isFinite(retryAfterSeconds) && retryAfterSeconds > 0) {
return new Date(Date.now() + retryAfterSeconds * 1000)
}
services/libs/connectors/src/http/client.ts:118
- Axios has no request timeout by default. A provider that accepts a connection but stops responding can therefore occupy an activity slot and pooled token until the 30-minute Temporal timeout. Apply a finite client default while still allowing connector-specific overrides.
return await axios.request<T>({ ...applyToken(config, token), validateStatus: () => true })
services/apps/connectors_worker/src/activities/syncRunActivities.ts:66
- Every connector failure is recorded as
framework.internal, even though this PR already introducesConnectorError.errorClassvalues for authentication, rate limits, provider failures, and contract errors. This makes persisted failure telemetry inaccurate and prevents class-specific handling; preserveConnectorError.errorClassand useunknownonly for unclassified exceptions.
})
services/libs/connectors/src/credentials.ts:9
- This raw integration-table query places database access inside the connector library, coupling connector logic directly to schema details and bypassing the shared data-access layer. Move the lookup into
services/libs/data-access-layerand keep only credential construction here.
const integration: { platform: string } | null = await qx.selectOneOrNone(
`SELECT platform
FROM integrations
WHERE id = $(integrationId) AND "deletedAt" IS NULL`,
services/libs/connectors/src/types.ts:14
- This comment only records POC status and a future design plan, so it will become stale without documenting a runtime invariant. Move the roadmap detail to CM-1372 and let the current type express the supported variant.
// POC only: single variant; becomes a discriminated union (token, oauth2, ...)
// as more connectors land
services/libs/connectors/src/credentials.ts:17
- This milestone note does not explain a runtime constraint and will become stale as connector support changes. Remove it and track the planned platform work in CM-1372.
// POC scope: GitHub only; each migrated connector adds its platform case here
services/libs/connectors/src/credentials.ts:28
- This comment describes temporary implementation status and a future replacement rather than a non-obvious invariant. Track the secret-manager migration in a ticket and remove the roadmap note from the code.
// POC only: secrets come from env; secret-manager adoption (OCI Vault) swaps this
// body without touching callers — getCredential stays the single entry point
services/apps/connectors_worker/src/main.ts:33
- This comment is a milestone roadmap rather than documentation of behavior the code must preserve. Move the M4 plan to CM-1372 and remove it from the implementation.
// POC only: dummy connector drives the control-plane end-to-end; real
// connectors register here starting with GitHub in M4
services/apps/connectors_worker/src/activities/syncRunActivities.ts:38
- This comment only describes the current POC limitation and planned M2 work, so it will become stale. Track the sink implementation in the ticket and remove this implementation-status note.
// POC only: emit collects counts in memory; the Kafka sink emitter arrives in M2
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 29 out of 30 changed files in this pull request and generated 1 comment.
Files not reviewed (1)
- pnpm-lock.yaml: Generated file
Suppressed comments (10)
services/apps/connectors_worker/src/activities/syncRunActivities.ts:72
- The HTTP layer introduced in this PR already exposes
ConnectorError.errorClass, but this catch records every failure asframework.internal, which is not even a member ofErrorClass. Auth, rate-limit, provider, and connector failures will therefore be stored inaccurately and dead-letter diagnostics cannot distinguish them. Persist the connector error class with anErrorClass-compatible fallback.
// POC only: everything unclassified is framework.internal; the 7-class
// error taxonomy arrives with the M2 HTTP client
await recordRunFailure(qx, unitId, 'framework.internal', DEAD_LETTER_AFTER)
scripts/services/docker/Dockerfile.connectors_worker:14
- This production stage uses Node.js 20, which has been EOL since April 30, 2026 and no longer receives security fixes. Deploy the worker on a supported LTS image such as Node.js 24.
FROM node:20-bookworm-slim as runner
services/libs/connectors/src/credentials.ts:28
- This comment describes a future secret-manager migration rather than a constraint callers must preserve. Remove it and track the migration in the relevant ticket instead.
// POC only: secrets come from env; secret-manager adoption (OCI Vault) swaps this
// body without touching callers — getCredential stays the single entry point
services/libs/connectors/src/http/client.ts:168
Retry-Aftermay legally be an HTTP date, not only a delay in seconds.Number()rejects that form and parks the token for the one-minute fallback, potentially resuming requests well before the provider permits them. Parse the date form before falling back.
}
const resetEpochSeconds = Number(headers['x-ratelimit-reset'])
if (Number.isFinite(resetEpochSeconds) && resetEpochSeconds > 0) {
return new Date(resetEpochSeconds * 1000)
scripts/services/docker/Dockerfile.connectors_worker:1
- Node.js 20 reached end of security support on April 30, 2026. Building this new service on an EOL runtime leaves the build environment without security fixes; use a currently supported LTS image such as Node.js 24 and keep both stages on the same supported major.
This issue also appears on line 14 of the same file.
FROM node:20-alpine as builder
services/libs/connectors/src/types.ts:14
- This comment only describes POC scope and a planned future type change; it does not document a non-obvious invariant. Remove it and let the current
Credentialtype express its supported variant.
// POC only: single variant; becomes a discriminated union (token, oauth2, ...)
// as more connectors land
services/libs/connectors/src/credentials.ts:17
- This comment records temporary scope and future work rather than a required invariant. Remove it; the switch already makes the currently supported platforms explicit.
This issue also appears on line 27 of the same file.
// POC scope: GitHub only; each migrated connector adds its platform case here
services/apps/connectors_worker/src/main.ts:33
- This comment is a note about the POC and a future milestone, while
registerConnector(dummyConnector)is already self-explanatory. Remove the comment and track the future registration work in its ticket.
// POC only: dummy connector drives the control-plane end-to-end; real
// connectors register here starting with GitHub in M4
services/apps/connectors_worker/src/activities/syncRunActivities.ts:38
- This comment only narrates the temporary POC implementation and future sink work. Remove it and track the Kafka emitter work in the relevant ticket.
This issue also appears on line 70 of the same file.
// POC only: emit collects counts in memory; the Kafka sink emitter arrives in M2
services/libs/connectors/src/credentials.ts:10
- This introduces a database query in the connector framework even though connector persistence is centralized in the data-access layer (see
services/libs/data-access-layer/src/connectors/syncUnits.ts:9). Move the integration lookup into that DAL module and keep this function responsible for credential construction; otherwise database access becomes split across packages and harder to evolve consistently.
const integration: { platform: string } | null = await qx.selectOneOrNone(
`SELECT platform
FROM integrations
WHERE id = $(integrationId) AND "deletedAt" IS NULL`,
{ integrationId },
| let lastError: ConnectorError = new ProviderUnavailableError() | ||
| for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) { | ||
| try { | ||
| return await attemptRequest<T>(deps, config, true) |
There was a problem hiding this comment.
Accepted trade-off: method-based retry restriction would break the primary use case — GitHub GraphQL sends read-only queries as POSTs. Connector syncs are read-only against the provider by design (writes go to our own sink via Kafka, never through this client), so retry-on-unavailable does not risk duplicated provider side effects here.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 29 out of 30 changed files in this pull request and generated 2 comments.
Files not reviewed (1)
- pnpm-lock.yaml: Generated file
Suppressed comments (9)
services/libs/connectors/src/http/client.ts:56
- This retry loop applies to every HTTP method. A timed-out or 5xx POST/PATCH may already have been committed by the provider, so replaying it can duplicate mutations. Restrict automatic retries to idempotent methods, or require an explicit retry opt-in/idempotency key for mutation requests.
for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
try {
return await attemptRequest<T>(deps, config, true)
services/libs/connectors/src/http/client.ts:128
- When a caller supplies
Authorization, the pooled token acquired above is not used, but a 401/403 still quarantines that unrelated token and a rate-limit response still parks it. Either always apply the acquired token in the default applier or reject pre-authenticated configs so token-pool state is never mutated for credentials that were not sent.
if (!headers.has('Authorization')) {
headers.set('Authorization', `Bearer ${token.value}`)
}
services/libs/connectors/src/http/client.ts:165
Retry-Aftermay be an HTTP-date as well as delta-seconds.Number(...)ignores the date form, so the token can be resumed after the 60-second fallback even when the provider requested a longer wait. Parse a future HTTP-date before falling back.
const retryAfterSeconds = Number(headers['retry-after'])
if (Number.isFinite(retryAfterSeconds) && retryAfterSeconds > 0) {
return new Date(Date.now() + retryAfterSeconds * 1000)
}
services/apps/connectors_worker/src/activities/syncRunActivities.ts:72
- Every connector failure is persisted as
framework.internal, including theConnectorErrorclasses introduced by this PR. This discardsprovider.rate_limit,provider.auth, and other classifications and makes those failures count toward dead-lettering as internal faults. Persisterr.errorClassforConnectorErrorinstances and use a valid fallback such asunknown.
// POC only: everything unclassified is framework.internal; the 7-class
// error taxonomy arrives with the M2 HTTP client
await recordRunFailure(qx, unitId, 'framework.internal', DEAD_LETTER_AFTER)
services/libs/connectors/src/http/client.ts:118
- Axios defaults to no request timeout, and this wrapper does not add one. A stalled provider connection can therefore retain a socket and activity resources beyond the Temporal start-to-close timeout because Temporal cannot interrupt the underlying Axios request. Apply a finite default timeout while still allowing callers to override it.
This issue also appears on line 162 of the same file.
return await axios.request<T>({ ...applyToken(config, token), validateStatus: () => true })
services/libs/connectors/src/types.ts:14
- This comment only records POC scope and planned type evolution; the current literal type already expresses the implementation. Remove the change note and track future connector variants in the relevant ticket.
// POC only: single variant; becomes a discriminated union (token, oauth2, ...)
// as more connectors land
services/libs/connectors/src/credentials.ts:17
- This is a scope/change note rather than a non-obvious invariant, and the switch already makes the supported platforms clear. Remove it; future platform work should be tracked in the ticket.
// POC scope: GitHub only; each migrated connector adds its platform case here
services/libs/connectors/src/credentials.ts:28
- This comment describes planned secret-manager work rather than a necessary invariant or external workaround. Remove it, or convert the work into a ticketed
TODO(CM-XXX):if it must remain visible in code.
// POC only: secrets come from env; secret-manager adoption (OCI Vault) swaps this
// body without touching callers — getCredential stays the single entry point
services/apps/connectors_worker/src/activities/syncRunActivities.ts:38
- This comment is a change note about a future sink implementation; the in-memory counter is already self-explanatory. Remove it and keep milestone planning in the ticket.
// POC only: emit collects counts in memory; the Kafka sink emitter arrives in M2
| // POC only: dummy connector drives the control-plane end-to-end; real | ||
| // connectors register here starting with GitHub in M4 | ||
| registerConnector(dummyConnector) |
There was a problem hiding this comment.
Accepted trade-off for the POC: units are seeded via manual SQL (how the M1 exit run was verified end-to-end, and how the M2 toy connector will run). The real bootstrap — discover() → upsertSyncUnits propagation wired into integration mutation paths — is explicitly M4 scope per the milestone plan.
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 29 out of 30 changed files in this pull request and generated 1 comment.
Files not reviewed (1)
- pnpm-lock.yaml: Generated file
Suppressed comments (12)
services/libs/connectors/src/credentials.ts:28
- This comment describes the current implementation and a planned replacement rather than a required invariant. Remove the change note;
githubAppCredentialand the environment-variable reads already explain the behavior.
// POC only: secrets come from env; secret-manager adoption (OCI Vault) swaps this
// body without touching callers — getCredential stays the single entry point
services/libs/data-access-layer/src/connectors/syncUnits.ts:58
- This claim result is sent through Temporal on every dispatcher tick, but the workflow only needs
id,platform, andsyncName;su.*also serializes the watermark and all run metadata for up to 100 units. Large JSONB watermarks will unnecessarily inflate activity payloads and workflow histories. Return a dedicated lightweight claim type with only those three columns.
RETURNING su.*`,
services/libs/connectors/src/http/client.ts:56
- This retry loop replays every request configuration, including non-idempotent
POST/PATCHcalls. If the provider commits a mutation and the response is lost or returns a retryable 5xx, the client can execute that mutation up to three times. Restrict retries to idempotent requests, or require callers to opt in with an idempotency key/retry-safe flag.
async function requestWithRetry<T>(deps: HttpClientDeps, config: AxiosRequestConfig): Promise<T> {
let lastError: ConnectorError = new ProviderUnavailableError()
for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
try {
return await attemptRequest<T>(deps, config, true)
services/libs/data-access-layer/src/connectors/syncUnits.ts:88
- A successful run resets the failure count but leaves the previous
lastErrorClassin place, so persisted state continues to report an error after recovery. Clear it as part of the same success update.
"consecutiveFailures" = 0,
"updatedAt" = now()
services/libs/connectors/src/http/client.ts:169
Retry-Aftermay be either delay-seconds or an HTTP date. Date-form values currently fall through to the 60-second fallback, which can unpark a token before the provider's requested time and trigger another rate limit.
function computeResumeAt(headers: Record<string, string>): Date {
const retryAfterSeconds = Number(headers['retry-after'])
if (Number.isFinite(retryAfterSeconds) && retryAfterSeconds > 0) {
return new Date(Date.now() + retryAfterSeconds * 1000)
}
const resetEpochSeconds = Number(headers['x-ratelimit-reset'])
if (Number.isFinite(resetEpochSeconds) && resetEpochSeconds > 0) {
return new Date(resetEpochSeconds * 1000)
}
return new Date(Date.now() + RATE_LIMIT_FALLBACK_MS)
services/libs/connectors/src/types.ts:14
- This future-shape note will become stale as credential variants are added and does not document a required invariant. Remove it; the single literal
kindalready makes the current limitation clear.
// POC only: single variant; becomes a discriminated union (token, oauth2, ...)
// as more connectors land
services/apps/connectors_worker/src/main.ts:33
- This milestone note only describes temporary registration state and will become stale when the first real connector lands. Remove it; the
registerConnector(dummyConnector)call is self-explanatory.
// POC only: dummy connector drives the control-plane end-to-end; real
// connectors register here starting with GitHub in M4
services/apps/connectors_worker/src/activities/syncRunActivities.ts:72
- The HTTP client introduced in this PR already emits typed
ConnectorErrorclasses, but this path persists every one as the non-taxonomy valueframework.internal. As a result,lastErrorClasscannot distinguish auth, rate-limit, contract, or availability failures, and all transient provider failures are counted toward the same dead-letter threshold. Preserve the connector error class and apply the intended retry/dead-letter policy for retryable classes.
} catch (err) {
log.error(err, 'sync run failed')
// POC only: everything unclassified is framework.internal; the 7-class
// error taxonomy arrives with the M2 HTTP client
await recordRunFailure(qx, unitId, 'framework.internal', DEAD_LETTER_AFTER)
services/libs/connectors/src/credentials.ts:17
- This POC/future-work note does not document an invariant and will become stale when discovery is implemented. Remove it and keep the platform cases self-explanatory.
This issue also appears on line 27 of the same file.
// POC scope: GitHub only; each migrated connector adds its platform case here
services/apps/connectors_worker/src/activities/syncRunActivities.ts:38
- This comment describes the temporary emitter implementation and future work rather than a required invariant. Remove it; the
emitcallback clearly shows that it only updates the count.
// POC only: emit collects counts in memory; the Kafka sink emitter arrives in M2
services/apps/connectors_worker/src/schedules/dispatcher.ts:29
- Wrapping the caught value in a new
Errordiscards the original error type, stack, and Temporal client metadata (and can reduce non-string values to[object Object]). Re-throw the original error as the other schedule registration paths do (services/apps/packages_worker/src/pypi/schedule.ts:37).
throw new Error(err)
services/libs/data-access-layer/src/connectors/syncUnits.ts:117
SELECT *couples this DAL contract to every future table column and fetches fields the sync activity never uses. Select the explicit fields required to execute a run and return a dedicated run-context type so schema additions are not silently pulled into worker memory.
`SELECT *
FROM integration.sync_units
WHERE id = $(id)`,
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 29 out of 30 changed files in this pull request and generated no new comments.
Files not reviewed (1)
- pnpm-lock.yaml: Generated file
Suppressed comments (10)
services/apps/connectors_worker/src/activities/syncRunActivities.ts:72
- The HTTP error taxonomy is already introduced in this PR, but every sync failure is persisted as
framework.internal, which is not even anErrorClass. Provider rate-limit, auth, contract, and availability failures therefore become indistinguishable and all count toward dead-lettering. PreserveConnectorError.errorClassand apply the intended retry/reschedule policy for retryable classes; useunknownonly for unclassified errors.
// POC only: everything unclassified is framework.internal; the 7-class
// error taxonomy arrives with the M2 HTTP client
await recordRunFailure(qx, unitId, 'framework.internal', DEAD_LETTER_AFTER)
services/libs/connectors/src/credentials.ts:28
- This secret-manager roadmap note is not an implementation invariant and can become stale. Remove it; the function name and environment-variable validation already explain the current behavior.
// POC only: secrets come from env; secret-manager adoption (OCI Vault) swaps this
// body without touching callers — getCredential stays the single entry point
services/libs/connectors/src/http/client.ts:32
- Budget correction is not tied to the token selected for this request. With concurrent pooled requests, an implementation cannot determine which token these response headers belong to, so remaining/reset values can be applied to the wrong token. Pass the acquired
token.idtogether with the headers.
correctBudget: (headers: Record<string, string>) => Promise<void>
services/libs/connectors/src/http/client.ts:174
Retry-Afteralso permits an HTTP-date, butNumber(...)rejects that valid form and parks the token for only the 60-second fallback. If the advertised date is later, the token is reused too early and receives repeated 429s. Parse both delay-seconds and HTTP-date forms before falling back.
const retryAfterSeconds = Number(headers['retry-after'])
if (Number.isFinite(retryAfterSeconds) && retryAfterSeconds > 0) {
return new Date(Date.now() + retryAfterSeconds * 1000)
}
services/libs/connectors/src/credentials.ts:9
- This package performs an integration-table query directly and imports the DAL's internal
src/queryExecutorpath. Integration SQL is otherwise centralized inservices/libs/data-access-layer/src/integrations/index.ts; keeping this query here crosses the data-access boundary and couples the connector API to DAL internals. Move the lookup to the DAL and pass its typed result into credential resolution.
const integration: { platform: string } | null = await qx.selectOneOrNone(
`SELECT platform
FROM integrations
WHERE id = $(integrationId) AND "deletedAt" IS NULL`,
services/libs/connectors/src/types.ts:15
- This forward-looking POC note only describes a planned type change and will become stale. The current interface already expresses the supported credential shape, so remove the note.
// POC only: single variant; becomes a discriminated union (token, oauth2, ...)
// as more connectors land
services/libs/connectors/src/credentials.ts:18
- This comment narrates temporary scope and future work rather than a non-obvious invariant. The switch is self-explanatory; remove the note.
This issue also appears on line 27 of the same file.
// POC scope: GitHub only; each migrated connector adds its platform case here
services/apps/connectors_worker/src/main.ts:34
- This milestone note records temporary implementation history rather than a constraint required to understand the registration. Remove it and let the explicit dummy registration describe the current setup.
// POC only: dummy connector drives the control-plane end-to-end; real
// connectors register here starting with GitHub in M4
services/apps/connectors_worker/src/activities/syncRunActivities.ts:39
- This comment only describes the current stub and future work. Remove it;
emittedCountand theemitcallback make the behavior clear.
This issue also appears on line 70 of the same file.
// POC only: emit collects counts in memory; the Kafka sink emitter arrives in M2
services/libs/connectors/src/http/client.ts:57
- The new retry and token-rotation state machine has no automated coverage, despite similar HTTP clients testing retry behavior (for example,
services/apps/packages_worker/src/blast-radius/crates/__tests__/registryClient.test.ts:106). Add deterministic tests for retry exhaustion, token rotation/parking, authentication quarantine, timeout mapping, and header-derived resume times.
async function requestWithRetry<T>(deps: HttpClientDeps, config: AxiosRequestConfig): Promise<T> {
let lastError: ConnectorError = new ProviderUnavailableError()
for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
try {
return await attemptRequest<T>(deps, config, true)
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 30 out of 31 changed files in this pull request and generated no new comments.
Files not reviewed (1)
- pnpm-lock.yaml: Generated file
Suppressed comments (13)
services/apps/connectors_worker/src/activities/syncRunActivities.ts:72
- The HTTP layer already exposes typed
ConnectorError.errorClassvalues, but every failure is stored asframework.internal, which is not even in that taxonomy. This also counts expectedprovider.rate_limitresponses toward the dead-letter threshold, so five rate-limit-only runs permanently disable a valid unit. Preserve the actual error class and reschedule rate limits usingoptions.resumeAtinstead of recording them as consecutive failures.
// POC only: everything unclassified is framework.internal; the 7-class
// error taxonomy arrives with the M2 HTTP client
await recordRunFailure(qx, unitId, 'framework.internal', DEAD_LETTER_AFTER)
services/libs/connectors/src/credentials.ts:17
- This comment only restates the switch and describes future connector work. The codebase guideline requires the code to be self-explanatory; remove the roadmap note or track required future work with a ticketed
TODO(CM-XXX).
// POC scope: GitHub only; each migrated connector adds its platform case here
services/libs/connectors/src/credentials.ts:28
- This is an unticketed implementation roadmap rather than a non-obvious invariant. Remove it, or use a two-line
TODO(CM-XXX):only if secret-manager migration must be tracked in code.
// POC only: secrets come from env; secret-manager adoption (OCI Vault) swaps this
// body without touching callers — getCredential stays the single entry point
services/libs/connectors/src/http/client.ts:96
- After the first 429 this recursive call disables rotation, so a pool with three or more tokens stops after the second rate-limited token and never tries the remaining healthy tokens. Continue rotating while newly acquired token IDs remain; pool exhaustion should be determined by
acquireToken, not by a one-rotation boolean.
return attemptRequest<T>(deps, config, false)
services/libs/connectors/src/http/client.ts:126
applyTokenexecutes inside this catch block, so an exception in connector token application is converted toprovider.unavailableand retried three times as though the provider failed. Apply the token before entering the Axios-only try/catch so connector/configuration errors retain their identity and are not retried.
...applyToken(config, token),
services/libs/connectors/src/http/client.ts:174
Retry-Afteralso permits an HTTP-date, but converting that form withNumberproducesNaNand parks the token for only the 60-second fallback. Providers can therefore be retried before their requested time. Parse a valid future HTTP-date before falling back; likewise ignore stale reset epochs.
const retryAfterSeconds = Number(headers['retry-after'])
if (Number.isFinite(retryAfterSeconds) && retryAfterSeconds > 0) {
return new Date(Date.now() + retryAfterSeconds * 1000)
}
services/libs/connectors/src/pool/tokenPool.ts:83
- Token selection and the LRU score update are separate Redis operations. Concurrent activities can all read the same oldest ID before any score changes and then send every request with that token, defeating load distribution and prematurely rate-limiting it while other tokens remain idle. Make selection/reservation atomic, for example with a Lua script or an equivalent Redis transaction.
const ordered = await redis.zRange(lruKey, 0, -1)
for (const id of ordered) {
const state = states.get(id)
if (state && isHealthy(state, nowMs)) {
await redis.zAdd(lruKey, { score: nowMs, value: id })
services/libs/connectors/src/types.ts:15
- This roadmap note is not a required invariant and the declaration already shows the current single variant. Remove it; future union work should be tracked as
TODO(CM-XXX)only when the ticket must remain visible in code.
// POC only: single variant; becomes a discriminated union (token, oauth2, ...)
// as more connectors land
services/apps/connectors_worker/src/main.ts:34
- This comment documents the current change and a future milestone rather than an invariant. Remove it; registration is self-explanatory, and planned connector work belongs in a ticketed
TODO(CM-XXX)if it must remain in code.
// POC only: dummy connector drives the control-plane end-to-end; real
// connectors register here starting with GitHub in M4
services/apps/connectors_worker/src/activities/syncRunActivities.ts:39
- This is an unticketed note about incomplete future sink work, which the codebase comment guideline disallows. Remove it or convert it to a two-line
TODO(CM-XXX):tied to the Kafka sink ticket.
This issue also appears on line 70 of the same file.
// POC only: emit collects counts in memory; the Kafka sink emitter arrives in M2
services/libs/connectors/src/credentials.ts:11
- This duplicates an integration lookup inside the connector library even though database access is centralized in the DAL and
services/libs/data-access-layer/src/integrations/index.ts:288already providesfetchIntegrationByIdwith the same soft-delete filter. Reuse that function so query behavior and integration typing do not diverge.
This issue also appears in the following locations of the same file:
- line 17
- line 27
const integration: { platform: string } | null = await qx.selectOneOrNone(
`SELECT platform
FROM integrations
WHERE id = $(integrationId) AND "deletedAt" IS NULL`,
services/libs/data-access-layer/src/connectors/syncUnits.ts:117
SELECT *returnscreatedAt,updatedAt, and every future column even though the function promisesISyncUnit. Use an explicit projection matching that interface so schema additions do not silently expand this worker payload or expose fields the API does not declare.
`SELECT *
FROM integration.sync_units
WHERE id = $(id)`,
services/libs/connectors/src/http/client.ts:49
- This new retry/rate-limit state machine has no automated coverage, despite analogous provider clients being tested (for example
services/apps/packages_worker/src/go/__tests__/proxyClient.test.tsandblast-radius/crates/__tests__/registryClient.test.ts). Add focused tests for unavailable retries, pools with more than two rate-limited tokens, auth quarantine, and both legalRetry-Afterformats.
export function createHttpClient(deps: HttpClientDeps): ConnectorHttp {
return {
request: <T>(config: AxiosRequestConfig) => requestWithRetry<T>(deps, config),
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 30 out of 31 changed files in this pull request and generated 2 comments.
Files not reviewed (1)
- pnpm-lock.yaml: Generated file
Suppressed comments (4)
services/libs/connectors/src/http/client.ts:105
- An authentication failure quarantines the bad token but immediately fails the request, even when another pooled token is healthy. This makes one stale credential fail the whole sync, unlike the rate-limit path that rotates tokens. Retry once with a fresh token after quarantine when token rotation is still allowed.
if (error.errorClass === 'provider.auth') {
await deps.quarantineToken(token.id)
deps.log.warn(
{ tokenId: token.id, status: response.status },
'token quarantined on auth failure',
services/libs/connectors/src/http/client.ts:32
- Budget state is per pooled token, but this callback receives only response headers. With concurrent requests, an implementation cannot reliably determine which token's bucket to correct, so it can update the wrong budget. Pass the acquired
token.idtocorrectBudgetand expose a matching per-token correction operation on the pool.
This issue also appears on line 101 of the same file.
correctBudget: (headers: Record<string, string>) => Promise<void>
services/apps/connectors_worker/src/activities/syncRunActivities.ts:72
- The error taxonomy and HTTP client are already introduced in this PR, yet every failure is persisted as
framework.internal, which is not even anErrorClassvalue. Provider auth, rate-limit, contract, and sink failures therefore lose their classification. PersistConnectorError.errorClasswhen available and useunknownfor unclassified errors.
// POC only: everything unclassified is framework.internal; the 7-class
// error taxonomy arrives with the M2 HTTP client
await recordRunFailure(qx, unitId, 'framework.internal', DEAD_LETTER_AFTER)
services/libs/connectors/src/pool/tokenPool.ts:199
- An empty pool has no available request headroom, but this returns
true. Once a probe is wired, integrations with no seeded tokens will be admitted and then fail inacquire()withtoken pool empty; report no headroom so they are deferred instead.
if (states.size === 0) {
return true
| const pool = createTokenPool(svc.redis, unit.platform, unit.integrationId) | ||
| if (await pool.hasHeadroom(DEFAULT_RUN_ESTIMATE)) { |
There was a problem hiding this comment.
Intentional for this milestone: Task 3 deliberately ships the admission machinery inert — probeBudget is the single source of truth for budgets and its GitHub implementation (GET /rate_limit) is explicitly M4 scope, where the wiring supplies the probe and the shared-pool connectionId resolver. Platforms without a probe are deliberately unbudgeted (admission bypass), not misconfigured. Removing and re-adding the admission path would just churn the diff.
| if (probe) { | ||
| const bucket = await loadBucket(probe, id, nowMs) | ||
| if (bucket && bucket.remaining <= 0) { | ||
| const resetAt = new Date(bucket.resetAtMs) | ||
| if (!earliestBudgetResetAt || resetAt < earliestBudgetResetAt) { | ||
| earliestBudgetResetAt = resetAt | ||
| } | ||
| continue | ||
| } | ||
| if (bucket) { | ||
| await redis.hIncrBy(bucketKey(id), 'remaining', -1) |
There was a problem hiding this comment.
Accepted trade-off for the POC (same class as the state read-modify-write noted in the code): budget counters are only a short-horizon burst guard — the probe re-snapshots every bucket to provider truth within ~90s, so a concurrent check/decrement race overshoots by at most a request or two, which the reactive rate-limit path (park + resumeAt) absorbs. An atomic Lua select+debit(+LRU) is the documented productization fix; not worth the embedded-Lua complexity at POC scale.
WIP