feat: extract cdn upload to outside the transaction - #2939
Closed
wilsonrivera wants to merge 2 commits into
Closed
Conversation
Contributor
WalkthroughThis PR introduces CompositionBlobStorageQueue and routes blob/router-config writes and deletions through it; CompositionService now accepts the queue and enqueues operations instead of calling blobStorage/composer uploads directly. Composer no longer receives repository instances; it constructs repositories internally using private fields. Handlers and repository call-sites instantiate and drain the queue after transactions. ChangesQueue-based blob storage and Composer repository encapsulation refactor
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (4)
controlplane/src/core/services/CompositionService.ts (1)
559-567:⚠️ Potential issue | 🟠 Major | ⚡ Quick winSurface mapper upload failures instead of swallowing them.
updateMapperForFederatedGraph()now enqueuesmapper.jsonwrites, butCompositionBlobStorageQueue.processQueue()currently ignoresupload-blobexceptions (controlplane/src/core/services/CompositionBlobStorageQueue.ts:94-98). That means a failed mapper upload is reported as a successful deployment even though the DB hashes were already updated and routers may read stale manifests.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controlplane/src/core/services/CompositionService.ts` around lines 559 - 567, The mapper upload failures are being swallowed by CompositionBlobStorageQueue.processQueue, so calls from updateMapperForFederatedGraph (which uses this.blobStorageQueue.enqueueBlobUpload) appear to succeed even when the underlying 'upload-blob' operation fails; modify processQueue's 'upload-blob' handler to stop catching and discarding exceptions — instead log the error with context and rethrow (or propagate) the exception so enqueueBlobUpload/ updateMapperForFederatedGraph can await the result and abort the deployment on failure; ensure enqueueBlobUpload returns a Promise that resolves/rejects with the underlying upload result and that updateMapperForFederatedGraph awaits it so failed uploads surface to callers.controlplane/src/core/bufservices/federated-graph/moveFederatedGraph.ts (1)
145-172:⚠️ Potential issue | 🟠 Major | ⚡ Quick winDrain the blob queue after the transaction commits.
Line 172 still waits for
cbsq.processQueue()inside theopts.db.transaction(...)callback, so the CDN/router-config work keeps the transaction open across external I/O. Hoist the queue out of the callback and process it only after the transaction resolves, otherwise this path still holds locks/connections for the full upload duration.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controlplane/src/core/bufservices/federated-graph/moveFederatedGraph.ts` around lines 145 - 172, The code currently calls cbsq.processQueue() inside the opts.db.transaction(...) callback which keeps the DB transaction open during external I/O; move the call to cbsq.processQueue() out of the transaction so queue processing runs after the transaction resolves. Concretely, keep constructing CompositionBlobStorageQueue (cbsq) and calling compositionService.composeAndDeployFederatedGraph(...) inside the transaction but do not await cbsq.processQueue() there; instead return any needed metadata from the transaction and, after the transaction promise resolves, call await cbsq.processQueue() and push its results into deploymentErrors so the CDN/uploads run after commit. Ensure cbsq remains in scope (hoisted) so it’s available post-transaction.controlplane/src/core/bufservices/contract/createContract.ts (1)
186-213:⚠️ Potential issue | 🟠 Major | ⚡ Quick winMove queue creation and draining out of the transaction.
Line 212 still awaits
cbsq.processQueue()before the transaction callback returns, so the CDN/admission work is still happening under the open DB transaction. In this file the queue is also built withtx, which hard-binds the post-commit path to the transaction object. Hoist the queue to the outer scope, construct it withopts.db, and only drain it afteropts.db.transaction(...)resolves.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controlplane/src/core/bufservices/contract/createContract.ts` around lines 186 - 213, Hoist construction of CompositionBlobStorageQueue out of the DB transaction: create the cbsq before calling opts.db.transaction(...) and instantiate it with opts.db (not tx) and the same other params (logger, opts.blobStorage, authContext.organizationId, { cdnBaseUrl: opts.cdnBaseUrl, webhookJWTSecret: opts.admissionWebhookJWTSecret }, opts.chClient, opts.webhookProxyUrl). Inside the transaction, continue to create CompositionService and call composeAndDeployFederatedGraph using the tx for DB operations but do NOT await cbsq.processQueue() there; instead collect deploymentErrors/compositionErrors/compositionWarnings from the transaction result and only call await cbsq.processQueue() after the opts.db.transaction(...) promise resolves so queue draining runs post-commit. Ensure any references to tx remain for DB actions and that CompositionService still uses tx for DB access while the queue uses opts.db.controlplane/src/core/repositories/FederatedGraphRepository.ts (1)
1451-1457:⚠️ Potential issue | 🔴 Critical | ⚡ Quick winUse the transaction client for
Composerin this transactional block.At Line 1456,
Composeris created withthis.dbeven though this logic runs insidethis.db.transaction(...). That lets Composer writes escape the current transaction and can leave persisted composition state even if the surrounding transaction fails.Suggested fix
- const composer = new Composer(this.logger, this.db, this.organizationId, chClient, webhookProxyUrl); + const composer = new Composer(this.logger, tx, this.organizationId, chClient, webhookProxyUrl);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controlplane/src/core/repositories/FederatedGraphRepository.ts` around lines 1451 - 1457, Inside the this.db.transaction(...) block the Composer is being instantiated with this.db which allows its writes to escape the transaction; change the Composer creation to use the transaction client (tx) instead of this.db so Composer's operations participate in the surrounding transaction (i.e., replace new Composer(this.logger, this.db, ...) with new Composer(this.logger, tx, ...) in FederatedGraphRepository where Composer is constructed).
🧹 Nitpick comments (4)
controlplane/src/core/blobstorage/index.ts (1)
23-29: ⚡ Quick winUse an
interfacefor this exported params contract.
PutObjectParamsis a public object shape, so it should follow the repo convention and be declared as aninterface.Suggested change
-export type PutObjectParams<Metadata extends Record<string, string>> = { +export interface PutObjectParams<Metadata extends Record<string, string>> { key: string; abortSignal?: AbortSignal; body: Buffer; contentType: string; metadata?: Metadata; -}; +}As per coding guidelines, "Prefer interfaces over type aliases for object shapes in TypeScript".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controlplane/src/core/blobstorage/index.ts` around lines 23 - 29, Replace the exported type alias PutObjectParams with an exported interface so the public params contract follows repo conventions; declare it as export interface PutObjectParams<Metadata extends Record<string, string>> { key: string; abortSignal?: AbortSignal; body: Buffer; contentType: string; metadata?: Metadata; } preserving the generic parameter name Metadata and all field names and optional markers exactly as in the current definition (use the same symbol PutObjectParams to avoid breaking references).Source: Coding guidelines
controlplane/src/core/composition/composer.ts (2)
142-157: ⚡ Quick winPrefer an
interfacefor this exported object shape.This is a public params contract, so keeping it as an
interfacewill match the repo’s TS conventions and make later extension/merging easier.Suggested change
-export type ComposeAndUploadRouterConfigParams = { +export interface ComposeAndUploadRouterConfigParams { admissionConfig: { jwtSecret: string; cdnBaseUrl: string; @@ federatedGraphAdmissionWebhookSecret?: string; actorId: string; pathOverride?: { ready: string; draft: string }; -}; +}As per coding guidelines, "Prefer interfaces over type aliases for object shapes in TypeScript".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controlplane/src/core/composition/composer.ts` around lines 142 - 157, Replace the exported type alias ComposeAndUploadRouterConfigParams with an exported interface of the same name: change "export type ComposeAndUploadRouterConfigParams = { ... }" to "export interface ComposeAndUploadRouterConfigParams { ... }" preserving all properties (admissionConfig, baseCompositionRouterExecutionConfig, baseCompositionSchemaVersionId, blobStorage, featureFlagRouterExecutionConfigByFeatureFlagName, federatedGraphId, organizationId, federatedGraphAdmissionWebhookURL, federatedGraphAdmissionWebhookSecret, actorId, pathOverride) and their types; keep the export name unchanged so existing references compile and allow future extension/merging per project convention.Source: Coding guidelines
386-398: ⚡ Quick winAdd an explicit return type to this public method.
This signature changed in the PR, so it is a good point to lock the returned shape down explicitly instead of relying on inference.
Suggested change
async composeAndUploadRouterConfig({ @@ federatedGraphAdmissionWebhookSecret, actorId, pathOverride, - }: ComposeAndUploadRouterConfigParams) { + }: ComposeAndUploadRouterConfigParams): Promise<{ errors: ComposeDeploymentError[] }> {As per coding guidelines, "Use explicit type annotations for function parameters and return types in TypeScript".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controlplane/src/core/composition/composer.ts` around lines 386 - 398, The public async function composeAndUploadRouterConfig currently relies on inferred return type; add an explicit return annotation (e.g. async composeAndUploadRouterConfig(...): Promise<RouterConfigUploadResult> { ... }) that represents the actual resolved shape, create or import a named interface/type (RouterConfigUploadResult or similarly descriptive name) that matches the function's returned object (status, url/id, any metadata or error info), and update any related exports/uses to consume that type so callers and the public API are type-locked instead of inferred.Source: Coding guidelines
controlplane/src/core/services/CompositionBlobStorageQueue.ts (1)
35-46: ⚡ Quick winAdd explicit
voidreturn types to enqueue methods.Please annotate these methods with
: voidfor consistency with the TS guideline.Suggested fix
- enqueueBlobUpload<Metadata extends Record<string, string>>(params: PutObjectParams<Metadata>) { + enqueueBlobUpload<Metadata extends Record<string, string>>(params: PutObjectParams<Metadata>): void { this.#queue.push({ action: 'upload-blob', params }); } - enqueueBlobDeletion(graph: FederatedGraphDTO, key: string) { + enqueueBlobDeletion(graph: FederatedGraphDTO, key: string): void { this.#queue.push({ action: 'delete-blob', graph, key }); } enqueueRouterConfigUpload( graph: FederatedGraphDTO, params: Omit<ComposeAndUploadRouterConfigParams, 'blobStorage' | 'admissionConfig'>, - ) { + ): void {As per coding guidelines,
**/*.{ts,tsx}should use explicit type annotations for function parameters and return types in TypeScript.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controlplane/src/core/services/CompositionBlobStorageQueue.ts` around lines 35 - 46, Add explicit ": void" return annotations to the enqueue methods to satisfy the TypeScript guideline: update enqueueBlobUpload<Metadata extends Record<string, string>>(params: PutObjectParams<Metadata>), enqueueBlobDeletion(graph: FederatedGraphDTO, key: string), and enqueueRouterConfigUpload(graph: FederatedGraphDTO, params: Omit<ComposeAndUploadRouterConfigParams, 'blobStorage' | 'admissionConfig'>) to declare their return type as void while leaving their bodies unchanged.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@controlplane/src/core/bufservices/federated-graph/migrateFromApollo.ts`:
- Line 187: The call to cbsq.processQueue() in migrateFromApollo is discarding
deployment errors from CompositionBlobStorageQueue.processQueue(), so migration
can report success even if deferred router-config/blob uploads failed; update
migrateFromApollo to capture the return value (or thrown errors) from
processQueue() and surface failures — e.g., if processQueue() returns an error
list or throws, log and propagate them (fail the migration by throwing or
returning an error result) so composeAndDeployFederatedGraph()'s deferred upload
errors do not get ignored; ensure you reference and handle
CompositionBlobStorageQueue.processQueue()'s result in the migrateFromApollo
flow.
In `@controlplane/src/core/bufservices/monograph/updateMonograph.ts`:
- Around line 140-149: The CompositionBlobStorageQueue is being drained via
cbsq.processQueue() while still inside the DB transaction and before
subgraphRepo.update(...), which causes actions enqueued during the update to be
missed and performs external side effects inside the transaction; change the
flow so subgraphRepo.update(...) (and any DB transaction commit) completes
first, then call cbsq.processQueue() outside the transaction and await its
result, capturing and handling any deployment errors returned (do not ignore
them), and ensure any webhook/CH calls from CompositionBlobStorageQueue happen
only after the DB transaction is committed.
In `@controlplane/src/core/composition/composer.ts`:
- Around line 166-177: The constructor currently instantiates transaction-scoped
repositories (`#federatedGraphRepo`, `#subgraphRepo`, `#contractRepo`,
`#graphCompositionRepository`) from the provided db which causes queued
composeAndUploadRouterConfig() runs to use a completed transaction; change the
implementation so Composer does not cache those repo instances: either store the
raw db/organizationId and lazily instantiate new repositories on demand via
private factory/getter methods (creating new FederatedGraphRepository,
SubgraphRepository, ContractRepository, GraphCompositionRepository each use) or
ensure the queued path (CompositionService) recreates a fresh Composer after
commit using a non-transactional db handle; update references to the cached
fields to use the new factory/getters or reconstructed Composer instead.
In `@controlplane/src/core/services/CompositionBlobStorageQueue.ts`:
- Around line 99-104: The upload failure in CompositionBlobStorageQueue is being
swallowed inside the 'upload-blob' case; update the handler that calls
this.blobStorage.putObject(entry.params) to catch the error and surface it
instead of ignoring it — e.g., log the error with context (including
entry.params identifiers) via the existing logger or processLogger and rethrow
or mark the queue entry as failed/put on a dead-letter queue so callers don't
assume success; locate the 'upload-blob' case in CompositionBlobStorageQueue and
change the empty catch to emit a clear error and propagate failure rather than
silently ignoring it.
---
Outside diff comments:
In `@controlplane/src/core/bufservices/contract/createContract.ts`:
- Around line 186-213: Hoist construction of CompositionBlobStorageQueue out of
the DB transaction: create the cbsq before calling opts.db.transaction(...) and
instantiate it with opts.db (not tx) and the same other params (logger,
opts.blobStorage, authContext.organizationId, { cdnBaseUrl: opts.cdnBaseUrl,
webhookJWTSecret: opts.admissionWebhookJWTSecret }, opts.chClient,
opts.webhookProxyUrl). Inside the transaction, continue to create
CompositionService and call composeAndDeployFederatedGraph using the tx for DB
operations but do NOT await cbsq.processQueue() there; instead collect
deploymentErrors/compositionErrors/compositionWarnings from the transaction
result and only call await cbsq.processQueue() after the
opts.db.transaction(...) promise resolves so queue draining runs post-commit.
Ensure any references to tx remain for DB actions and that CompositionService
still uses tx for DB access while the queue uses opts.db.
In `@controlplane/src/core/bufservices/federated-graph/moveFederatedGraph.ts`:
- Around line 145-172: The code currently calls cbsq.processQueue() inside the
opts.db.transaction(...) callback which keeps the DB transaction open during
external I/O; move the call to cbsq.processQueue() out of the transaction so
queue processing runs after the transaction resolves. Concretely, keep
constructing CompositionBlobStorageQueue (cbsq) and calling
compositionService.composeAndDeployFederatedGraph(...) inside the transaction
but do not await cbsq.processQueue() there; instead return any needed metadata
from the transaction and, after the transaction promise resolves, call await
cbsq.processQueue() and push its results into deploymentErrors so the
CDN/uploads run after commit. Ensure cbsq remains in scope (hoisted) so it’s
available post-transaction.
In `@controlplane/src/core/repositories/FederatedGraphRepository.ts`:
- Around line 1451-1457: Inside the this.db.transaction(...) block the Composer
is being instantiated with this.db which allows its writes to escape the
transaction; change the Composer creation to use the transaction client (tx)
instead of this.db so Composer's operations participate in the surrounding
transaction (i.e., replace new Composer(this.logger, this.db, ...) with new
Composer(this.logger, tx, ...) in FederatedGraphRepository where Composer is
constructed).
In `@controlplane/src/core/services/CompositionService.ts`:
- Around line 559-567: The mapper upload failures are being swallowed by
CompositionBlobStorageQueue.processQueue, so calls from
updateMapperForFederatedGraph (which uses
this.blobStorageQueue.enqueueBlobUpload) appear to succeed even when the
underlying 'upload-blob' operation fails; modify processQueue's 'upload-blob'
handler to stop catching and discarding exceptions — instead log the error with
context and rethrow (or propagate) the exception so enqueueBlobUpload/
updateMapperForFederatedGraph can await the result and abort the deployment on
failure; ensure enqueueBlobUpload returns a Promise that resolves/rejects with
the underlying upload result and that updateMapperForFederatedGraph awaits it so
failed uploads surface to callers.
---
Nitpick comments:
In `@controlplane/src/core/blobstorage/index.ts`:
- Around line 23-29: Replace the exported type alias PutObjectParams with an
exported interface so the public params contract follows repo conventions;
declare it as export interface PutObjectParams<Metadata extends Record<string,
string>> { key: string; abortSignal?: AbortSignal; body: Buffer; contentType:
string; metadata?: Metadata; } preserving the generic parameter name Metadata
and all field names and optional markers exactly as in the current definition
(use the same symbol PutObjectParams to avoid breaking references).
In `@controlplane/src/core/composition/composer.ts`:
- Around line 142-157: Replace the exported type alias
ComposeAndUploadRouterConfigParams with an exported interface of the same name:
change "export type ComposeAndUploadRouterConfigParams = { ... }" to "export
interface ComposeAndUploadRouterConfigParams { ... }" preserving all properties
(admissionConfig, baseCompositionRouterExecutionConfig,
baseCompositionSchemaVersionId, blobStorage,
featureFlagRouterExecutionConfigByFeatureFlagName, federatedGraphId,
organizationId, federatedGraphAdmissionWebhookURL,
federatedGraphAdmissionWebhookSecret, actorId, pathOverride) and their types;
keep the export name unchanged so existing references compile and allow future
extension/merging per project convention.
- Around line 386-398: The public async function composeAndUploadRouterConfig
currently relies on inferred return type; add an explicit return annotation
(e.g. async composeAndUploadRouterConfig(...): Promise<RouterConfigUploadResult>
{ ... }) that represents the actual resolved shape, create or import a named
interface/type (RouterConfigUploadResult or similarly descriptive name) that
matches the function's returned object (status, url/id, any metadata or error
info), and update any related exports/uses to consume that type so callers and
the public API are type-locked instead of inferred.
In `@controlplane/src/core/services/CompositionBlobStorageQueue.ts`:
- Around line 35-46: Add explicit ": void" return annotations to the enqueue
methods to satisfy the TypeScript guideline: update enqueueBlobUpload<Metadata
extends Record<string, string>>(params: PutObjectParams<Metadata>),
enqueueBlobDeletion(graph: FederatedGraphDTO, key: string), and
enqueueRouterConfigUpload(graph: FederatedGraphDTO, params:
Omit<ComposeAndUploadRouterConfigParams, 'blobStorage' | 'admissionConfig'>) to
declare their return type as void while leaving their bodies unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 301eba46-f498-42ca-8789-2ce0fa1c9471
📒 Files selected for processing (30)
controlplane/src/core/blobstorage/index.tscontrolplane/src/core/bufservices/contract/createContract.tscontrolplane/src/core/bufservices/contract/updateContract.tscontrolplane/src/core/bufservices/feature-flag/createFeatureFlag.tscontrolplane/src/core/bufservices/feature-flag/deleteFeatureFlag.tscontrolplane/src/core/bufservices/feature-flag/enableFeatureFlag.tscontrolplane/src/core/bufservices/feature-flag/recomposeFeatureFlag.tscontrolplane/src/core/bufservices/feature-flag/updateFeatureFlag.tscontrolplane/src/core/bufservices/federated-graph/createFederatedGraph.tscontrolplane/src/core/bufservices/federated-graph/migrateFromApollo.tscontrolplane/src/core/bufservices/federated-graph/moveFederatedGraph.tscontrolplane/src/core/bufservices/federated-graph/updateFederatedGraph.tscontrolplane/src/core/bufservices/graph/recomposeGraph.tscontrolplane/src/core/bufservices/graph/setGraphRouterCompatibilityVersion.tscontrolplane/src/core/bufservices/monograph/publishMonograph.tscontrolplane/src/core/bufservices/monograph/updateMonograph.tscontrolplane/src/core/bufservices/proposal/createProposal.tscontrolplane/src/core/bufservices/proposal/updateProposal.tscontrolplane/src/core/bufservices/subgraph/deleteFederatedSubgraph.tscontrolplane/src/core/bufservices/subgraph/fixSubgraphSchema.tscontrolplane/src/core/bufservices/subgraph/moveSubgraph.tscontrolplane/src/core/bufservices/subgraph/publishFederatedSubgraph.tscontrolplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.tscontrolplane/src/core/bufservices/subgraph/updateSubgraph.tscontrolplane/src/core/composition/composeGraphs.worker.tscontrolplane/src/core/composition/composer.tscontrolplane/src/core/repositories/FederatedGraphRepository.tscontrolplane/src/core/repositories/SubgraphRepository.tscontrolplane/src/core/services/CompositionBlobStorageQueue.tscontrolplane/src/core/services/CompositionService.ts
| }; | ||
| } | ||
|
|
||
| await cbsq.processQueue(); |
Contributor
There was a problem hiding this comment.
Don't discard post-transaction deployment errors here.
processQueue() returns deployment errors from the deferred router-config/blob work (controlplane/src/core/services/CompositionBlobStorageQueue.ts:61-111), but this line ignores them. After moving uploads out of composeAndDeployFederatedGraph(), that means migration can still emit success audit/webhook flow and return OK even when the actual deployment failed.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@controlplane/src/core/bufservices/federated-graph/migrateFromApollo.ts` at
line 187, The call to cbsq.processQueue() in migrateFromApollo is discarding
deployment errors from CompositionBlobStorageQueue.processQueue(), so migration
can report success even if deferred router-config/blob uploads failed; update
migrateFromApollo to capture the return value (or thrown errors) from
processQueue() and surface failures — e.g., if processQueue() returns an error
list or throws, log and propagate them (fail the migration by throwing or
returning an error result) so composeAndDeployFederatedGraph()'s deferred upload
errors do not get ignored; ensure you reference and handle
CompositionBlobStorageQueue.processQueue()'s result in the migrateFromApollo
flow.
Contributor
Author
There was a problem hiding this comment.
We are not doing anything with the deployment errors here
Contributor
There was a problem hiding this comment.
@wilsonrivera, understood — MigrateFromApolloResponse only carries response (code/details) and token, so there's no field to surface deployment errors to the caller, unlike the composition-related handlers. That makes the intentional discard reasonable given the current proto contract.
One minor note: the errors are silently dropped without any trace. If the deferred uploads do fail, it could be hard to diagnose in production. It might be worth at least logging the result, e.g.:
const deploymentErrors = await cbsq.processQueue();
if (deploymentErrors.length > 0) {
logger.warn({ deploymentErrors }, 'Apollo migration completed but deferred uploads had errors');
}That said, I'll leave it to your discretion.
✏️ Learnings added
Learnt from: wilsonrivera
Repo: wundergraph/cosmo PR: 2939
File: controlplane/src/core/bufservices/federated-graph/migrateFromApollo.ts:187-187
Timestamp: 2026-06-09T12:56:28.385Z
Learning: In `controlplane/src/core/bufservices/federated-graph/migrateFromApollo.ts`, the result of `cbsq.processQueue()` is intentionally not captured or returned. This is because `MigrateFromApolloResponse` (proto) only has `response` (code/details) and `token` fields — it has no `deploymentErrors` field — so there is no mechanism to surface blob-storage queue errors to the caller. The design decision is to silently discard these errors for the migration flow.
Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.
🧠 Learnings used
Learnt from: thisisnithin
Repo: wundergraph/cosmo PR: 2771
File: controlplane/src/core/services/ApiGenerator.ts:4-7
Timestamp: 2026-04-16T18:53:03.303Z
Learning: In the cosmo controlplane codebase, the `traced` decorator is intentionally not applied to static methods when they are lightweight utility functions that perform no I/O. When reviewing `controlplane/**/*.ts`, do not flag missing tracing on static methods that are purely computational/helpers (e.g., building query conditions, generating patterns/keys) and do not call databases, network/HTTP, files, or other external side effects. If a static method does perform I/O or other observable external work, tracing omission should be flagged.
Learnt from: Aenimus
Repo: wundergraph/cosmo PR: 2860
File: controlplane/src/core/repositories/SubgraphRepository.ts:579-579
Timestamp: 2026-05-13T00:24:13.515Z
Learning: In the wundergraph/cosmo controlplane, the project targets Node.js 22+ and uses TypeScript with "lib": ["ESNext"] and "target": "esnext". Therefore, it is valid to use iterator helper methods (e.g., Iterator.prototype.map and related methods) on built-in iterators such as the iterators returned by Map.prototype.keys(), Map.prototype.values(), and Map.prototype.entries(). Do not flag usages of .map() or other iterator helper methods on these built-in iterators for runtime or TypeScript compatibility concerns in this codebase.
Comment on lines
+140
to
+149
| const cbsq = new CompositionBlobStorageQueue( | ||
| logger, | ||
| opts.db, | ||
| opts.blobStorage, | ||
| authContext.organizationId, | ||
| { cdnBaseUrl: opts.cdnBaseUrl, webhookJWTSecret: opts.admissionWebhookJWTSecret }, | ||
| opts.chClient, | ||
| opts.webhookProxyUrl, | ||
| ); | ||
|
|
Contributor
There was a problem hiding this comment.
Queue draining is placed at the wrong point and drops later queued work.
At Line 174, processQueue() runs before Line 177 (subgraphRepo.update(...)), so any actions enqueued during subgraph update are never processed. It also performs external side effects inside the DB transaction and ignores returned deployment errors.
Suggested flow adjustment
- return opts.db.transaction(async (tx) => {
+ const cbsq = new CompositionBlobStorageQueue(
+ logger,
+ opts.db,
+ opts.blobStorage,
+ authContext.organizationId,
+ { cdnBaseUrl: opts.cdnBaseUrl, webhookJWTSecret: opts.admissionWebhookJWTSecret },
+ opts.chClient,
+ opts.webhookProxyUrl,
+ );
+
+ const result = await opts.db.transaction(async (tx) => {
...
- const cbsq = new CompositionBlobStorageQueue(...);
const compositionService = new CompositionService(..., cbsq, ...);
await fedGraphRepo.update(...);
- await cbsq.processQueue();
await subgraphRepo.update(..., compositionService);
...
return {
response: { code: EnumStatusCode.OK },
compositionErrors: [],
};
});
+
+ const deploymentErrors = await cbsq.processQueue();
+ // handle deploymentErrors (map to response/ERR_DEPLOYMENT_FAILED or raise)
+ return result;Also applies to: 174-175, 177-193
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@controlplane/src/core/bufservices/monograph/updateMonograph.ts` around lines
140 - 149, The CompositionBlobStorageQueue is being drained via
cbsq.processQueue() while still inside the DB transaction and before
subgraphRepo.update(...), which causes actions enqueued during the update to be
missed and performs external side effects inside the transaction; change the
flow so subgraphRepo.update(...) (and any DB transaction commit) completes
first, then call cbsq.processQueue() outside the transaction and await its
result, capturing and handling any deployment errors returned (do not ignore
them), and ensure any webhook/CH calls from CompositionBlobStorageQueue happen
only after the DB transaction is committed.
Comment on lines
+99
to
+104
| case 'upload-blob': { | ||
| try { | ||
| await this.blobStorage.putObject(entry.params); | ||
| } catch { | ||
| // ignore | ||
| } |
Contributor
There was a problem hiding this comment.
Do not silently swallow blob upload failures.
At Line 102, upload failures are ignored, so callers can return success while queued blob artifacts are missing.
Suggested fix
case 'upload-blob': {
try {
await this.blobStorage.putObject(entry.params);
- } catch {
- // ignore
+ } catch (err) {
+ this.logger.error({ err, key: entry.params.key }, 'Failed to upload blob');
+ if (err instanceof Error) {
+ errors.push({
+ message: `Failed to upload blob "${entry.params.key}": ${err.message}`,
+ });
+ }
}
break;
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| case 'upload-blob': { | |
| try { | |
| await this.blobStorage.putObject(entry.params); | |
| } catch { | |
| // ignore | |
| } | |
| case 'upload-blob': { | |
| try { | |
| await this.blobStorage.putObject(entry.params); | |
| } catch (err) { | |
| this.logger.error({ err, key: entry.params.key }, 'Failed to upload blob'); | |
| if (err instanceof Error) { | |
| errors.push({ | |
| message: `Failed to upload blob "${entry.params.key}": ${err.message}`, | |
| }); | |
| } | |
| } | |
| break; | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@controlplane/src/core/services/CompositionBlobStorageQueue.ts` around lines
99 - 104, The upload failure in CompositionBlobStorageQueue is being swallowed
inside the 'upload-blob' case; update the handler that calls
this.blobStorage.putObject(entry.params) to catch the error and surface it
instead of ignoring it — e.g., log the error with context (including
entry.params identifiers) via the existing logger or processLogger and rethrow
or mark the queue entry as failed/put on a dead-letter queue so callers don't
assume success; locate the 'upload-blob' case in CompositionBlobStorageQueue and
change the empty catch to emit a clear error and propagate failure rather than
silently ignoring it.
Codecov Report❌ Patch coverage is ❌ Your patch check has failed because the patch coverage (84.84%) is below the target coverage (90.00%). You can increase the patch coverage or adjust the target coverage. Additional details and impacted files@@ Coverage Diff @@
## main #2939 +/- ##
==========================================
+ Coverage 65.12% 65.25% +0.12%
==========================================
Files 327 328 +1
Lines 47130 47272 +142
Branches 5239 5256 +17
==========================================
+ Hits 30695 30846 +151
+ Misses 16411 16402 -9
Partials 24 24
🚀 New features to boost your workflow:
|
Contributor
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
controlplane/src/core/services/CompositionBlobStorageQueue.ts (1)
61-111:⚠️ Potential issue | 🔴 Critical | ⚡ Quick winQueue is never cleared after draining, causing duplicate processing on subsequent calls.
The
drainQueue()method iterates overthis.#queuebut never clears it. IfdrainQueue()is called multiple times (e.g., due to retry logic or code flow changes), all actions will be re-processed, leading to duplicate uploads/deletions.🐛 Proposed fix to clear the queue
async drainQueue(): Promise<PlainMessage<DeploymentError>[]> { const errors: PlainMessage<DeploymentError>[] = []; if (this.#queue.length === 0) { return errors; } + const actions = this.#queue; + this.#queue = []; + const composer = new Composer(this.logger, this.db, this.organizationId, this.chClient, this.webhookProxyUrl); - for (const entry of this.#queue) { + for (const entry of actions) { switch (entry.action) {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controlplane/src/core/services/CompositionBlobStorageQueue.ts` around lines 61 - 111, drainQueue() iterates this.#queue but never clears it, causing duplicate re-processing; modify drainQueue() to ensure the queue is cleared when processing begins or when processing completes (use a try/finally) so entries aren't re-run on subsequent calls—for example, capture the current batch (or set this.#queue = []) before/inside processing and place the clear in a finally block to guarantee cleanup even on errors; reference the drainQueue method and the private field this.#queue when making the change.
♻️ Duplicate comments (2)
controlplane/src/core/services/CompositionBlobStorageQueue.ts (1)
99-104:⚠️ Potential issue | 🟠 Major | ⚡ Quick winInconsistent error handling:
upload-bloblogs but doesn't report errors, unlikedelete-blob.The
delete-blobcase (lines 84-95) pushes errors to the returned array so callers can see deployment failures. Theupload-blobcase only logs, meaning callers will see success even when blob uploads fail. This asymmetry can cause data inconsistency where the DB transaction committed but artifacts are missing.If blob upload failures should be non-fatal (fire-and-forget), document this explicitly. Otherwise, align with
delete-blobbehavior.This was flagged in a past review—logging was added but errors still aren't surfaced to callers.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controlplane/src/core/services/CompositionBlobStorageQueue.ts` around lines 99 - 104, The upload-blob branch currently swallows errors by only calling this.logger.error in the case 'upload-blob' block where this.blobStorage.putObject(entry.params) fails; update it to surface failures the same way as the delete-blob path by pushing a failure object/error into the method's returned results array (or rethrow if the desired behavior is to fail fast) so callers see upload failures; reference the case 'upload-blob', the call to this.blobStorage.putObject(entry.params), and the logger.error invocation and mirror the error-pushing logic used in the case 'delete-blob' so behavior is consistent or explicitly document otherwise.controlplane/src/core/bufservices/monograph/updateMonograph.ts (1)
140-148:⚠️ Potential issue | 🔴 Critical | 🏗️ Heavy liftQueue is created and drained inside the transaction, defeating the PR's purpose.
Unlike all other handlers in this PR,
updateMonographcreatesCompositionBlobStorageQueueinside the DB transaction (lines 140-148) and drains it inside the transaction (line 193). This keeps CDN uploads within the transaction, which contradicts the PR objective to "extract CDN upload outside the transaction."This causes:
- Extended transaction duration due to external I/O
- Risk of transaction timeouts on slow CDN operations
- Potential DB inconsistency if CDN operations fail mid-transaction
Move
cbsqinstantiation before the transaction anddrainQueue()after it, matching the pattern in other handlers (deleteFeatureFlag,createFederatedGraph, etc.).🐛 Suggested structure
+ const cbsq = new CompositionBlobStorageQueue( + logger, + opts.db, + opts.blobStorage, + authContext.organizationId, + { cdnBaseUrl: opts.cdnBaseUrl, webhookJWTSecret: opts.admissionWebhookJWTSecret }, + opts.chClient, + opts.webhookProxyUrl, + ); + return opts.db.transaction(async (tx) => { // ... existing code ... - const cbsq = new CompositionBlobStorageQueue( - logger, - opts.db, - opts.blobStorage, - authContext.organizationId, - { cdnBaseUrl: opts.cdnBaseUrl, webhookJWTSecret: opts.admissionWebhookJWTSecret }, - opts.chClient, - opts.webhookProxyUrl, - ); // ... compositionService and updates ... - await cbsq.drainQueue(); // ... audit log and webhooks ... - return { response: { code: EnumStatusCode.OK }, compositionErrors: [] }; + return { compositionErrors: [], deploymentErrors: [] }; }); + + const queueErrors = await cbsq.drainQueue(); + // Handle queueErrors in response...Also applies to: 193-193
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controlplane/src/core/bufservices/monograph/updateMonograph.ts` around lines 140 - 148, The CompositionBlobStorageQueue is created and drained inside updateMonograph's DB transaction; instantiate the queue (new CompositionBlobStorageQueue(...)) before beginning the transaction and remove its usage from inside the transaction block, then call cbsq.drainQueue() only after the transaction has committed (or rolled back) so CDN uploads occur outside the transaction; update references in updateMonograph accordingly to mirror the pattern used by deleteFeatureFlag/createFederatedGraph (create queue before transaction, drainQueue() after).
🧹 Nitpick comments (1)
controlplane/src/core/services/CompositionBlobStorageQueue.ts (1)
103-103: 💤 Low valueUse template literal instead of string concatenation.
As per coding guidelines, prefer template literals over string concatenation.
♻️ Suggested fix
- this.logger.error(`Failed to upload blob "${entry.params.key}": ${err}`); + this.logger.error(`Failed to upload blob "${entry.params.key}": ${String(err)}`);Note: Using
String(err)makes the coercion explicit and handles non-Error objects cleanly.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@controlplane/src/core/services/CompositionBlobStorageQueue.ts` at line 103, Replace the string concatenation in the error log with a template literal and explicitly coerce the error to a string; locate the logger call inside CompositionBlobStorageQueue (the upload/failure handling block where this.logger.error is called for entry.params.key) and change the message to use a template literal like `Failed to upload blob "${entry.params.key}": ${String(err)}` so non-Error values are handled consistently.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@controlplane/src/core/services/CompositionBlobStorageQueue.ts`:
- Around line 61-111: drainQueue() iterates this.#queue but never clears it,
causing duplicate re-processing; modify drainQueue() to ensure the queue is
cleared when processing begins or when processing completes (use a try/finally)
so entries aren't re-run on subsequent calls—for example, capture the current
batch (or set this.#queue = []) before/inside processing and place the clear in
a finally block to guarantee cleanup even on errors; reference the drainQueue
method and the private field this.#queue when making the change.
---
Duplicate comments:
In `@controlplane/src/core/bufservices/monograph/updateMonograph.ts`:
- Around line 140-148: The CompositionBlobStorageQueue is created and drained
inside updateMonograph's DB transaction; instantiate the queue (new
CompositionBlobStorageQueue(...)) before beginning the transaction and remove
its usage from inside the transaction block, then call cbsq.drainQueue() only
after the transaction has committed (or rolled back) so CDN uploads occur
outside the transaction; update references in updateMonograph accordingly to
mirror the pattern used by deleteFeatureFlag/createFederatedGraph (create queue
before transaction, drainQueue() after).
In `@controlplane/src/core/services/CompositionBlobStorageQueue.ts`:
- Around line 99-104: The upload-blob branch currently swallows errors by only
calling this.logger.error in the case 'upload-blob' block where
this.blobStorage.putObject(entry.params) fails; update it to surface failures
the same way as the delete-blob path by pushing a failure object/error into the
method's returned results array (or rethrow if the desired behavior is to fail
fast) so callers see upload failures; reference the case 'upload-blob', the call
to this.blobStorage.putObject(entry.params), and the logger.error invocation and
mirror the error-pushing logic used in the case 'delete-blob' so behavior is
consistent or explicitly document otherwise.
---
Nitpick comments:
In `@controlplane/src/core/services/CompositionBlobStorageQueue.ts`:
- Line 103: Replace the string concatenation in the error log with a template
literal and explicitly coerce the error to a string; locate the logger call
inside CompositionBlobStorageQueue (the upload/failure handling block where
this.logger.error is called for entry.params.key) and change the message to use
a template literal like `Failed to upload blob "${entry.params.key}":
${String(err)}` so non-Error values are handled consistently.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: df00fd5c-7d32-43f1-83a3-392fd8e1b4bc
📒 Files selected for processing (21)
controlplane/src/core/bufservices/contract/createContract.tscontrolplane/src/core/bufservices/contract/updateContract.tscontrolplane/src/core/bufservices/feature-flag/createFeatureFlag.tscontrolplane/src/core/bufservices/feature-flag/deleteFeatureFlag.tscontrolplane/src/core/bufservices/feature-flag/enableFeatureFlag.tscontrolplane/src/core/bufservices/feature-flag/recomposeFeatureFlag.tscontrolplane/src/core/bufservices/feature-flag/updateFeatureFlag.tscontrolplane/src/core/bufservices/federated-graph/createFederatedGraph.tscontrolplane/src/core/bufservices/federated-graph/migrateFromApollo.tscontrolplane/src/core/bufservices/federated-graph/moveFederatedGraph.tscontrolplane/src/core/bufservices/federated-graph/updateFederatedGraph.tscontrolplane/src/core/bufservices/graph/recomposeGraph.tscontrolplane/src/core/bufservices/graph/setGraphRouterCompatibilityVersion.tscontrolplane/src/core/bufservices/monograph/publishMonograph.tscontrolplane/src/core/bufservices/monograph/updateMonograph.tscontrolplane/src/core/bufservices/subgraph/deleteFederatedSubgraph.tscontrolplane/src/core/bufservices/subgraph/moveSubgraph.tscontrolplane/src/core/bufservices/subgraph/publishFederatedSubgraph.tscontrolplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.tscontrolplane/src/core/bufservices/subgraph/updateSubgraph.tscontrolplane/src/core/services/CompositionBlobStorageQueue.ts
🚧 Files skipped from review as they are similar to previous changes (14)
- controlplane/src/core/bufservices/contract/updateContract.ts
- controlplane/src/core/bufservices/subgraph/publishFederatedSubgraphs.ts
- controlplane/src/core/bufservices/feature-flag/recomposeFeatureFlag.ts
- controlplane/src/core/bufservices/graph/recomposeGraph.ts
- controlplane/src/core/bufservices/federated-graph/updateFederatedGraph.ts
- controlplane/src/core/bufservices/graph/setGraphRouterCompatibilityVersion.ts
- controlplane/src/core/bufservices/feature-flag/createFeatureFlag.ts
- controlplane/src/core/bufservices/contract/createContract.ts
- controlplane/src/core/bufservices/federated-graph/moveFederatedGraph.ts
- controlplane/src/core/bufservices/subgraph/publishFederatedSubgraph.ts
- controlplane/src/core/bufservices/federated-graph/migrateFromApollo.ts
- controlplane/src/core/bufservices/subgraph/moveSubgraph.ts
- controlplane/src/core/bufservices/monograph/publishMonograph.ts
- controlplane/src/core/bufservices/feature-flag/updateFeatureFlag.ts
|
This PR was marked stale due to lack of activity. It will be closed in 14 days. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by CodeRabbit
Refactor
Bug Fixes
Checklist
Open Source AI Manifesto
This project follows the principles of the Open Source AI Manifesto. Please ensure your contribution aligns with its principles.