Skip to content

perf(vcs): run VCS sync batches in the placed fetch handler #1553

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Makisuo merged 3 commits into from
Oct 19, 2026
Merged

perf(vcs): run VCS sync batches in the placed fetch handler #1553

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b2695d7
perf(vcs): run VCS sync batches in the placed fetch handler
Makisuo Oct 19, 2026
0327869
fix(vcs): serve the batch route only to the self binding, and settle …
Makisuo Oct 19, 2026
15a8812
fix(vcs): authenticate the batch hop with a secret only the api Worke…
Makisuo Oct 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 5 additions & 24 deletions apps/alerting/src/tick-dispatch.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,10 @@
* the database and the warehouse. The fire hands its tick to this Worker's own fetch handler
* over a self service binding instead, and that invocation is placed.
*/
import { internalTokenFor, selfFetcher } from "@maple/backend/platform/self-binding"
import { isValidInternalBearer } from "@maple/backend/services/auth/internal-auth"
import { workerEnvLayer } from "@maple/infra/worker-runtime"
import * as Cloudflare from "alchemy/Cloudflare"
import { Config, Effect, Option, Redacted, Schema } from "effect"
import { Effect, Option, Schema } from "effect"
import { HttpClient, HttpClientRequest } from "effect/http"
import { isLaneOf, type TickLane } from "./schedule.ts"

Expand Down Expand Up @@ -43,25 +43,6 @@ export class TickRequestRefused extends Schema.TaggedError<TickRequestRefused>()
},
) {}

const isFetcher = (value: unknown): value is Fetcher =>
typeof value === "object" && value !== null && "fetch" in value && typeof value.fetch === "function"

/** Surfaces a 3xx as the response instead of following it, so the bearer never leaves the hop. */
const withManualRedirects = (binding: Fetcher): Fetcher => ({
fetch: (input, init) => binding.fetch(input, { ...init, redirect: "manual" }),
connect: (address, options) => binding.connect(address, options),
})

/** The internal token as this invocation's env says; unreadable reads as unset. */
const internalTokenFor = (env: Record<string, unknown>) =>
Config.option(Config.Redacted("INTERNAL_SERVICE_TOKEN")).pipe(
Effect.map(Option.map(Redacted.value)),
Effect.orElseSucceed(() => Option.none<string>()),
// The invocation's env is the boundary this config belongs to.
// oxlint-disable-next-line effecttsgo/strict-effect-provide
Effect.provide(workerEnvLayer(env)),
)

/**
* Runs one lane of `cron` in the placed fetch handler. Fails with `TickDispatchUnavailable` when it
* did not run there (no binding, no token, or the handler refused it), so the caller runs it inline.
Expand All @@ -72,9 +53,9 @@ export const dispatchTick = Effect.fn("alerting.dispatch_tick")(function* (
env: Record<string, unknown>,
) {
yield* Effect.annotateCurrentSpan("maple.alerting.lane", lane)
const binding = env[SELF_BINDING]
const binding = selfFetcher(env, SELF_BINDING)
const token = yield* internalTokenFor(env)
if (!isFetcher(binding)) {
if (Option.isNone(binding)) {
return yield* new TickDispatchUnavailable({
message: "No self service binding",
reason: "no-binding",
Expand All @@ -87,7 +68,7 @@ export const dispatchTick = Effect.fn("alerting.dispatch_tick")(function* (
})
}

const client = Cloudflare.toHttpClient(Cloudflare.fromCloudflareFetcher(withManualRedirects(binding)))
const client = Cloudflare.toHttpClient(Cloudflare.fromCloudflareFetcher(binding.value))
const request = HttpClientRequest.post(TICK_URL).pipe(
HttpClientRequest.setUrlParam("cron", cron),
HttpClientRequest.setUrlParam("lane", lane),
Expand Down
4 changes: 4 additions & 0 deletions apps/api/src/http/api-observability.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import { Layer } from "effect"
import { Headers, HttpMiddleware } from "effect/http"
import { VCS_SYNC_BATCH_PATH } from "../worker/vcs-sync-dispatch"

// OAuth callbacks whose query string carries a provider-issued authorization
// `code` (exchangeable for an access token) plus the single-use connect `state`.
Expand Down Expand Up @@ -30,6 +31,9 @@ export const ApiObservabilityLive = Layer.mergeAll(
HttpMiddleware.TracerDisabledWhen,
(request: { url: string; method: string }) =>
request.url === "/health" ||
// A VCS sync batch: its spans are `maple-vcs-sync`'s, and a minute-long server span here
// would skew this service's latency.
request.url === VCS_SYNC_BATCH_PATH ||
request.method === "OPTIONS" ||
OAUTH_CALLBACK_PATH.test(request.url) ||
EMAIL_UNSUBSCRIBE_PATH.test(request.url) ||
Expand Down
209 changes: 124 additions & 85 deletions apps/api/src/vcs-sync-runtime.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
import { EdgeCacheServiceLive } from "@maple/backend/platform/CacheBackendLive"
import { eventTelemetry } from "@maple/infra/worker-telemetry"
import { Cause, Effect, Layer, Option } from "effect"
import { EventBaseLive } from "@maple/backend/platform/DatabasePgLive"

Expand All @@ -25,18 +24,24 @@ import {
import { CiTraceSinkLive } from "@maple/backend/services/integrations/ci-traces/CiTraceService"
import { summarizeCause } from "@maple/backend/platform/describe-cause"
import type { QueueBatch } from "@maple/backend/platform/queue-batch"
import type { Message } from "@cloudflare/workers-types"
import {
applyOutcome,
type MessageOutcome,
type PlacedMessage,
toPlacedMessage,
VCS_SYNC_MAX_RETRIES,
vcsSyncTelemetry,
} from "./worker/vcs-sync-dispatch"

// Per-invocation runtime for the VCS sync queue, independent of the HTTP graph.
// No tracer or logger of its own: the Worker provides `vcsSyncTelemetry` around
// the event, and a layer here that carried one would shadow it. The Worker env,
// its `ConfigProvider` and the binding ports come from the Worker too
// (`apiPorts`), provided around the event.

/**
* Deliberately not `maple-api`: background work sharing the request-facing
* service's name skewed its percentiles (p99 32s, 2026-09-04).
*/
export const vcsSyncTelemetry = eventTelemetry({ serviceName: "maple-vcs-sync" })
// Defined beside the dispatch, which the Worker imports eagerly; one SDK instance per isolate.
export { vcsSyncTelemetry }

// One delivery, two readers: the issue link / verification window, and the
// review trigger. Each is isolated in the fan-out so a defect in
Expand Down Expand Up @@ -99,100 +104,134 @@ export const runScheduledSync = Effect.gen(function* () {
Effect.withSpan("VcsScheduledSync.tick"),
)

// Must match the consumer's `maxRetries` in worker.ts. No DLQ exists, so on the
// final delivery (attempt > max_retries) we persist a terminal status instead of silently dropping.
const VCS_SYNC_MAX_RETRIES = 3
// No DLQ exists, so on the final delivery (attempt > max_retries) we persist a terminal status
// instead of silently dropping.

const settleMessage = (service: VcsSyncService["Service"], message: PlacedMessage) =>
service.processMessage(message.body).pipe(
Effect.matchCauseEffect({
onFailure: (cause): Effect.Effect<MessageOutcome> => {
// Rate-limited: delay redelivery until the VCS budget resets instead of retrying immediately.
const failure = Option.getOrUndefined(Cause.findErrorOption(cause))
const isRateLimited = failure?._tag === "@maple/http/errors/VcsRateLimitedError"

const delaySeconds = isRateLimited
? clampQueueDelaySeconds(failure.retryAfterSeconds)
: undefined
const isDelaySecondsSet = delaySeconds !== undefined
// Last retry exhausted: persist terminal status so repos don't get stuck backfilling, then ack.
const isFinalAttempt = message.attempts > VCS_SYNC_MAX_RETRIES

// Low-cardinality outcome label; the full Cause stays in the log, not the span.
const outcome = isFinalAttempt ? "exhausted" : isDelaySecondsSet ? "retry_delayed" : "retry"

return Effect.annotateCurrentSpan({
"vcs.queue.message.outcome": outcome,
// Tag rate-limit errors so `retry_delayed`/`exhausted` spans are filterable without parsing logs.
...(isRateLimited
? { "vcs.queue.failure.tag": "@maple/http/errors/VcsRateLimitedError" }
: undefined),
...(isDelaySecondsSet ? { "vcs.queue.retry.delay_seconds": delaySeconds } : undefined),
}).pipe(
Effect.flatMap(() =>
Effect.logError("[VCS] sync message failed").pipe(
Effect.annotateLogs({
error: summarizeCause(cause),
attempt: message.attempts,
outcome,
...(isFinalAttempt ? { exhausted: true } : undefined),
...(isDelaySecondsSet ? { retryDelaySeconds: delaySeconds } : undefined),
}),
),
),
Effect.flatMap((): Effect.Effect<MessageOutcome> =>
isFinalAttempt
? service
.recordExhaustedFailure(message.body)
.pipe(Effect.as({ id: message.id, outcome: "ack" }))
: Effect.succeed(
isDelaySecondsSet
? { id: message.id, outcome: "retry", delaySeconds }
: { id: message.id, outcome: "retry" },
),
),
)
},
onSuccess: () =>
Effect.annotateCurrentSpan({ "vcs.queue.message.outcome": "succeeded_ack" }).pipe(
Effect.as<MessageOutcome>({ id: message.id, outcome: "ack" }),
),
}),
Effect.withSpan("VcsSyncQueue.processMessage", {
kind: "consumer",
attributes: {
"messaging.system": MESSAGING_SYSTEM,
"messaging.destination.name": MESSAGING_DESTINATION,
"messaging.operation.name": "process",
"messaging.message.delivery_attempt": message.attempts,
},
}),
)

const batchSpan = (messageCount: number, dispatch: "inline" | "placed") =>
Effect.withSpan("VcsSyncQueue.processBatch", {
kind: "consumer",
attributes: {
"messaging.system": MESSAGING_SYSTEM,
"messaging.destination.name": MESSAGING_DESTINATION,
"messaging.operation.name": "receive",
"messaging.batch.message_count": messageCount,
"maple.vcs.dispatch": dispatch,
},
})

/** The batch in this invocation, each message acked or retried as soon as it settles. */
export const processBatch = (batch: QueueBatch) =>
Effect.gen(function* () {
const service = yield* VcsSyncService
yield* Effect.forEach(
batch.messages,
(message) =>
service.processMessage(message.body).pipe(
Effect.matchCauseEffect({
onFailure: (cause) => {
// Rate-limited: delay redelivery until the VCS budget resets instead of retrying immediately.
const failure = Option.getOrUndefined(Cause.findErrorOption(cause))
const isRateLimited = failure?._tag === "@maple/http/errors/VcsRateLimitedError"

const delaySeconds = isRateLimited
? clampQueueDelaySeconds(failure.retryAfterSeconds)
: undefined
const isDelaySecondsSet = delaySeconds !== undefined
// Last retry exhausted: persist terminal status so repos don't get stuck backfilling, then ack.
const isFinalAttempt = message.attempts > VCS_SYNC_MAX_RETRIES

// Low-cardinality outcome label — full Cause stays in the log, not the span.
const outcome = isFinalAttempt
? "exhausted"
: isDelaySecondsSet
? "retry_delayed"
: "retry"

return Effect.annotateCurrentSpan({
"vcs.queue.message.outcome": outcome,
// Tag rate-limit errors so `retry_delayed`/`exhausted` spans are filterable without parsing logs.
...(isRateLimited
? {
"vcs.queue.failure.tag": "@maple/http/errors/VcsRateLimitedError",
}
: undefined),
...(isDelaySecondsSet
? { "vcs.queue.retry.delay_seconds": delaySeconds }
: undefined),
}).pipe(
Effect.flatMap(() =>
Effect.logError("[VCS] sync message failed").pipe(
Effect.annotateLogs({
error: summarizeCause(cause),
attempt: message.attempts,
outcome,
...(isFinalAttempt ? { exhausted: true } : undefined),
...(isDelaySecondsSet
? { retryDelaySeconds: delaySeconds }
: undefined),
}),
),
),
Effect.flatMap(() =>
isFinalAttempt
? service
.recordExhaustedFailure(message.body)
.pipe(Effect.flatMap(() => Effect.sync(() => message.ack())))
: Effect.sync(() =>
isDelaySecondsSet
? message.retry({ delaySeconds })
: message.retry(),
),
),
)
},
onSuccess: () =>
Effect.annotateCurrentSpan({
"vcs.queue.message.outcome": "succeeded_ack",
}).pipe(Effect.flatMap(() => Effect.sync(() => message.ack()))),
}),
Effect.withSpan("VcsSyncQueue.processMessage", {
kind: "consumer",
attributes: {
"messaging.system": MESSAGING_SYSTEM,
"messaging.destination.name": MESSAGING_DESTINATION,
"messaging.operation.name": "process",
"messaging.message.delivery_attempt": message.attempts,
},
}),
settleMessage(service, toPlacedMessage(message)).pipe(
Effect.flatMap((outcome) => applyOutcome(message, outcome)),
),
{ discard: true },
)
}).pipe(batchSpan(batch.messages.length, "inline"))

/** The batch in the placed fetch handler: the outcomes go back to the consumer that holds the messages. */
export const processPlacedBatch = (messages: ReadonlyArray<PlacedMessage>, placement: string) =>
Effect.gen(function* () {
const service = yield* VcsSyncService
yield* Effect.annotateCurrentSpan("maple.placement", placement)
return yield* Effect.forEach(messages, (message) => settleMessage(service, message))
}).pipe(batchSpan(messages.length, "placed"))

/**
* Final deliveries whose placed run failed with an unknown outcome: no redelivery is left, so the
* terminal status is recorded and the message acked, as a failed final attempt inline would be.
*/
export const recordExhaustedBatch = (messages: ReadonlyArray<Message<unknown>>) =>
Effect.gen(function* () {
const service = yield* VcsSyncService
yield* Effect.forEach(
messages,
(message) =>
Effect.logError("[VCS] sync message exhausted after a failed placed run").pipe(
Effect.annotateLogs({ attempt: message.attempts, outcome: "exhausted", exhausted: true }),
Effect.andThen(service.recordExhaustedFailure(message.body)),
Effect.andThen(Effect.sync(() => message.ack())),
),
{ discard: true },
)
}).pipe(
Effect.withSpan("VcsSyncQueue.processBatch", {
Effect.withSpan("VcsSyncQueue.recordExhausted", {
kind: "consumer",
attributes: {
"messaging.system": MESSAGING_SYSTEM,
"messaging.destination.name": MESSAGING_DESTINATION,
"messaging.operation.name": "receive",
"messaging.batch.message_count": batch.messages.length,
"messaging.operation.name": "settle",
"messaging.batch.message_count": messages.length,
},
}),
)
14 changes: 11 additions & 3 deletions apps/api/src/worker.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import { IsolateAge } from "@maple/infra/isolate-age"
import { isolateContext } from "@maple/infra/worker-http"
import { WorkerTelemetry } from "@maple/infra/worker-telemetry"
import * as Cloudflare from "alchemy/Cloudflare"
import { makeRandom } from "alchemy/Random"
import * as AlchemyTelemetry from "alchemy/Telemetry"
import { Effect, Layer } from "effect"
import { ApiObservabilityLive } from "./http/api-observability"
Expand All @@ -17,6 +18,8 @@ import { ApiBindingsLive, apiPorts, bindApiClients } from "./worker/bindings"
import { registerQueueConsumers } from "./worker/consumers"
import { registerCrons } from "./worker/crons"
import { makeAppGraphs, makeFetch } from "./worker/http"
import { vcsSyncBatchRoute } from "./worker/vcs-sync"
import { SELF_BINDING, VCS_SYNC_HOP_SECRET } from "./worker/vcs-sync-dispatch"
import ClickHouseSchemaApplyWorkflow from "./workflows/ClickHouseSchemaApplyWorkflow"

/** `__ALCHEMY_RUNTIME__` folds to `true` in the bundle, so the stack-side branch is tree-shaken. */
Expand All @@ -41,6 +44,10 @@ const props = Effect.gen(function* () {
env: {
...mapleDbEnv(db, "api"),
AI_WORKER: ai,
// VCS sync batches hop through it so they run where placement applies.
[SELF_BINDING]: Cloudflare.Workers.Self,
// Minted once and kept in state; bound to this Worker alone.
[VCS_SYNC_HOP_SECRET]: yield* makeRandom("VcsSyncHopSecret"),
...configuredEnv,
...devEnv,
},
Expand All @@ -54,14 +61,15 @@ export default class MapleApi extends Cloudflare.Worker<MapleApi>()(
// Yielding a hosted class binds, registers and exports it; do not move it to another Worker.
const schemaApply = yield* ClickHouseSchemaApplyWorkflow
const clients = yield* bindApiClients
const ports = apiPorts(clients, schemaApply, yield* Cloudflare.WorkerEnvironment)
const env = yield* Cloudflare.WorkerEnvironment
const ports = apiPorts(clients, schemaApply, env)
// Graphs build on the first event, not here: init also runs at plan time, where
// alchemy would auto-bind every `Config` read. Lazy build keeps `/health` answering.
const isolate = isolateContext(yield* Effect.context())
const { app, queryApp } = yield* makeAppGraphs(isolate, ports)
yield* registerCrons(ports)
yield* registerQueueConsumers(ports)
return { fetch: yield* makeFetch(app, ports, queryApp) }
yield* registerQueueConsumers(ports, env)
return { fetch: yield* makeFetch(app, ports, queryApp, vcsSyncBatchRoute(ports, env)) }
}).pipe(
// The init is the entry point: cron and queue sources need the host Worker.
// oxlint-disable-next-line effecttsgo/strict-effect-provide
Expand Down
Loading
Loading