feat: optimizations - #175
Merged
Merged
Conversation
Two issues surfaced by the load-test stack added in 0dff7a5: WIP-22: wire_rpc_server_duration_seconds is a single histogram shared by unary and streaming RPCs. WatchCommands handler returns at worker disconnect (session lifetime, seconds-to-hours), so its sample lands in the +Inf bucket of a histogram capped at 10 s. histogram_quantile then pins p99 at exactly 10 s. Doc lays out the kind=unary|streaming attribute fix. WIP-23: Submit path is lock+fsync-bound. Two sequential pebble.Sync writes per submit (job meta + config), the first held under c.mu, so worker Heartbeat / UpdateTaskStatus queue behind disk I/O — observed 9.9 s tail. Doc proposes batched WriteBatch, persist-outside-lock, and optional NoSync for non-load-bearing writes. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
recordRPCServerOp now takes a kind ("unary" | "streaming") and skips the
histogram record for streaming methods. Their "duration" is the stream
lifetime (the worker session — minutes to hours), so the sample lands in
the +Inf bucket of the 10s-capped histogram and pins p99 at exactly 10s.
The kind label is added to the count + error counters too so dashboards
can still see "stream opened" events distinct from unary calls.
Updated dashboard panels 30/31/32 (RPC requests, p99 latency, error rate)
to filter {kind="unary"}; future streaming methods inherit the right
behaviour without dashboard changes.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…23.1)
SubmitJob was doing two sequential pebble.Sync writes — JobMetaKey under
c.mu, then JobConfigKey after unlock. Each write paid one fsync (~6ms
floor), and at sustained submit load the second commit queued behind
other goroutines' commits.
Replace with a single store.WriteBatch([]KVPair{meta, config}) under the
lock. Same atomicity guarantee (recovery treats meta+config as a unit
already), one fsync instead of two, and the unlock now happens after a
single disk operation rather than two consecutive ones.
Race tests pass.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Previously SubmitJob held c.mu.Lock() across the WriteBatch fsync,
which made every concurrent worker RPC (Heartbeat, UpdateTaskStatus)
queue behind disk I/O. Observed Heartbeat p99 of 9.89s and
UpdateTaskStatus p99 of 9.78s under sustained submit load.
Restructure to:
1. Take c.mu.Lock briefly to dup-check the name and reserve
c.jobs[job.ID] = job.
2. Release the lock.
3. Persist via store.WriteBatch outside the lock.
4. On persist failure, take the lock again to roll back the
reservation.
The reservation is what serialises concurrent same-name submits —
the duplicate check sees the already-inserted entry and rejects.
Trade-off explicitly accepted in WIP-23: a crash between the in-memory
insert and the disk commit drops the job. c.jobs is rebuilt from Pebble
on restart, so the submitter sees a 201 but the job is gone — same
failure mode as a network-partitioned ACK and recoverable by client
retry. recovery_test.go continues to exercise the rebuild path.
Race tests pass.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
🔒 Security Scan ResultsFound 26 issues (🔴 18 high · 🟠 8 medium · 🟡 0 low). Findings
…and 1 more. Full SARIF report on the repo's Security tab. |
tarungka
added a commit
that referenced
this pull request
May 9, 2026
TaskSlot.Run launches the operator chain alongside output writers and watermark goroutines under one errgroup. The chain's wrapper has `defer runCancel()` so peers exit when the chain finishes. When an operator panics: 1. Chain recovers, sets retErr = ErrOperatorPanic. 2. Chain wrapper's defers run: runCancel() fires. 3. Peer goroutines see gctx.Done(), return context.Canceled. 4. Chain's wrapper returns retErr to errgroup. 5. errgroup picks the FIRST non-nil err via errOnce.Do — race. If a peer wins step 5 with context.Canceled, errgroup.Wait() returns context.Canceled, which TaskSlot.Run filtered to nil — masking the panic entirely. Under -race the perturbation makes peers win ~1 in 50 runs; flaky CI on PR #175 surfaced it. Fix: capture the chain's terminal error in atomic.Pointer[error] *before* runCancel() fires, prefer it over errgroup's verdict in Run(). Guarded by !errors.Is(*e, context.Canceled) so we don't mask errors from peer goroutines (e.g. source reader) when the chain itself bowed out via gctx cancellation — TestTaskSlot_SourceReadError_FailsTask exercises that path. Verification: go test -race -count=200 -run TestTaskSlot_OperatorPanic ./internal/engine/... → 200 consecutive passes (was ~1 fail in 50 under -race). Full repo `go test -race ./...` green. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
tarungka
added a commit
that referenced
this pull request
May 9, 2026
* docs(trds): WIP-22 + WIP-23 deferred fix plans Two issues surfaced by the load-test stack added in 0dff7a5: WIP-22: wire_rpc_server_duration_seconds is a single histogram shared by unary and streaming RPCs. WatchCommands handler returns at worker disconnect (session lifetime, seconds-to-hours), so its sample lands in the +Inf bucket of a histogram capped at 10 s. histogram_quantile then pins p99 at exactly 10 s. Doc lays out the kind=unary|streaming attribute fix. WIP-23: Submit path is lock+fsync-bound. Two sequential pebble.Sync writes per submit (job meta + config), the first held under c.mu, so worker Heartbeat / UpdateTaskStatus queue behind disk I/O — observed 9.9 s tail. Doc proposes batched WriteBatch, persist-outside-lock, and optional NoSync for non-load-bearing writes. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * fix(rpc): exclude streaming RPCs from server-duration histogram (WIP-22) recordRPCServerOp now takes a kind ("unary" | "streaming") and skips the histogram record for streaming methods. Their "duration" is the stream lifetime (the worker session — minutes to hours), so the sample lands in the +Inf bucket of the 10s-capped histogram and pins p99 at exactly 10s. The kind label is added to the count + error counters too so dashboards can still see "stream opened" events distinct from unary calls. Updated dashboard panels 30/31/32 (RPC requests, p99 latency, error rate) to filter {kind="unary"}; future streaming methods inherit the right behaviour without dashboard changes. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * perf(coordinator): batch SubmitJob persists into one WriteBatch (WIP-23.1) SubmitJob was doing two sequential pebble.Sync writes — JobMetaKey under c.mu, then JobConfigKey after unlock. Each write paid one fsync (~6ms floor), and at sustained submit load the second commit queued behind other goroutines' commits. Replace with a single store.WriteBatch([]KVPair{meta, config}) under the lock. Same atomicity guarantee (recovery treats meta+config as a unit already), one fsync instead of two, and the unlock now happens after a single disk operation rather than two consecutive ones. Race tests pass. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * perf(coordinator): persist job outside c.mu critical section (WIP-23.2) Previously SubmitJob held c.mu.Lock() across the WriteBatch fsync, which made every concurrent worker RPC (Heartbeat, UpdateTaskStatus) queue behind disk I/O. Observed Heartbeat p99 of 9.89s and UpdateTaskStatus p99 of 9.78s under sustained submit load. Restructure to: 1. Take c.mu.Lock briefly to dup-check the name and reserve c.jobs[job.ID] = job. 2. Release the lock. 3. Persist via store.WriteBatch outside the lock. 4. On persist failure, take the lock again to roll back the reservation. The reservation is what serialises concurrent same-name submits — the duplicate check sees the already-inserted entry and rejects. Trade-off explicitly accepted in WIP-23: a crash between the in-memory insert and the disk commit drops the job. c.jobs is rebuilt from Pebble on restart, so the submitter sees a 201 but the job is gone — same failure mode as a network-partitioned ACK and recoverable by client retry. recovery_test.go continues to exercise the rebuild path. Race tests pass. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * fix(engine): preserve operator-chain error past errgroup race (WIP-24) TaskSlot.Run launches the operator chain alongside output writers and watermark goroutines under one errgroup. The chain's wrapper has `defer runCancel()` so peers exit when the chain finishes. When an operator panics: 1. Chain recovers, sets retErr = ErrOperatorPanic. 2. Chain wrapper's defers run: runCancel() fires. 3. Peer goroutines see gctx.Done(), return context.Canceled. 4. Chain's wrapper returns retErr to errgroup. 5. errgroup picks the FIRST non-nil err via errOnce.Do — race. If a peer wins step 5 with context.Canceled, errgroup.Wait() returns context.Canceled, which TaskSlot.Run filtered to nil — masking the panic entirely. Under -race the perturbation makes peers win ~1 in 50 runs; flaky CI on PR #175 surfaced it. Fix: capture the chain's terminal error in atomic.Pointer[error] *before* runCancel() fires, prefer it over errgroup's verdict in Run(). Guarded by !errors.Is(*e, context.Canceled) so we don't mask errors from peer goroutines (e.g. source reader) when the chain itself bowed out via gctx cancellation — TestTaskSlot_SourceReadError_FailsTask exercises that path. Verification: go test -race -count=200 -run TestTaskSlot_OperatorPanic ./internal/engine/... → 200 consecutive passes (was ~1 fail in 50 under -race). Full repo `go test -race ./...` green. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * docs(trds): WIP-24 fix mermaid sequence-diagram parse errors Two issues in the original diagrams: - <br/> inside note text is fragile in sequenceDiagram (works in flowchart, breaks here when followed by special chars). - A note line beginning with "-race" was parsed as an arrow start because Mermaid sees "-" after ":" and tries to consume an arrow token, then fails when the next line continues with "Writer-->>". Rewrite all three diagrams without <br/> and without lines starting with a dash. Same content, just inlined or split into separate Note lines. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
tarungka
added a commit
that referenced
this pull request
May 10, 2026
* docs(trds): WIP-22 + WIP-23 deferred fix plans Two issues surfaced by the load-test stack added in 0dff7a5: WIP-22: wire_rpc_server_duration_seconds is a single histogram shared by unary and streaming RPCs. WatchCommands handler returns at worker disconnect (session lifetime, seconds-to-hours), so its sample lands in the +Inf bucket of a histogram capped at 10 s. histogram_quantile then pins p99 at exactly 10 s. Doc lays out the kind=unary|streaming attribute fix. WIP-23: Submit path is lock+fsync-bound. Two sequential pebble.Sync writes per submit (job meta + config), the first held under c.mu, so worker Heartbeat / UpdateTaskStatus queue behind disk I/O — observed 9.9 s tail. Doc proposes batched WriteBatch, persist-outside-lock, and optional NoSync for non-load-bearing writes. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * fix(rpc): exclude streaming RPCs from server-duration histogram (WIP-22) recordRPCServerOp now takes a kind ("unary" | "streaming") and skips the histogram record for streaming methods. Their "duration" is the stream lifetime (the worker session — minutes to hours), so the sample lands in the +Inf bucket of the 10s-capped histogram and pins p99 at exactly 10s. The kind label is added to the count + error counters too so dashboards can still see "stream opened" events distinct from unary calls. Updated dashboard panels 30/31/32 (RPC requests, p99 latency, error rate) to filter {kind="unary"}; future streaming methods inherit the right behaviour without dashboard changes. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * perf(coordinator): batch SubmitJob persists into one WriteBatch (WIP-23.1) SubmitJob was doing two sequential pebble.Sync writes — JobMetaKey under c.mu, then JobConfigKey after unlock. Each write paid one fsync (~6ms floor), and at sustained submit load the second commit queued behind other goroutines' commits. Replace with a single store.WriteBatch([]KVPair{meta, config}) under the lock. Same atomicity guarantee (recovery treats meta+config as a unit already), one fsync instead of two, and the unlock now happens after a single disk operation rather than two consecutive ones. Race tests pass. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * perf(coordinator): persist job outside c.mu critical section (WIP-23.2) Previously SubmitJob held c.mu.Lock() across the WriteBatch fsync, which made every concurrent worker RPC (Heartbeat, UpdateTaskStatus) queue behind disk I/O. Observed Heartbeat p99 of 9.89s and UpdateTaskStatus p99 of 9.78s under sustained submit load. Restructure to: 1. Take c.mu.Lock briefly to dup-check the name and reserve c.jobs[job.ID] = job. 2. Release the lock. 3. Persist via store.WriteBatch outside the lock. 4. On persist failure, take the lock again to roll back the reservation. The reservation is what serialises concurrent same-name submits — the duplicate check sees the already-inserted entry and rejects. Trade-off explicitly accepted in WIP-23: a crash between the in-memory insert and the disk commit drops the job. c.jobs is rebuilt from Pebble on restart, so the submitter sees a 201 but the job is gone — same failure mode as a network-partitioned ACK and recoverable by client retry. recovery_test.go continues to exercise the rebuild path. Race tests pass. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * fix(engine): preserve operator-chain error past errgroup race (WIP-24) TaskSlot.Run launches the operator chain alongside output writers and watermark goroutines under one errgroup. The chain's wrapper has `defer runCancel()` so peers exit when the chain finishes. When an operator panics: 1. Chain recovers, sets retErr = ErrOperatorPanic. 2. Chain wrapper's defers run: runCancel() fires. 3. Peer goroutines see gctx.Done(), return context.Canceled. 4. Chain's wrapper returns retErr to errgroup. 5. errgroup picks the FIRST non-nil err via errOnce.Do — race. If a peer wins step 5 with context.Canceled, errgroup.Wait() returns context.Canceled, which TaskSlot.Run filtered to nil — masking the panic entirely. Under -race the perturbation makes peers win ~1 in 50 runs; flaky CI on PR #175 surfaced it. Fix: capture the chain's terminal error in atomic.Pointer[error] *before* runCancel() fires, prefer it over errgroup's verdict in Run(). Guarded by !errors.Is(*e, context.Canceled) so we don't mask errors from peer goroutines (e.g. source reader) when the chain itself bowed out via gctx cancellation — TestTaskSlot_SourceReadError_FailsTask exercises that path. Verification: go test -race -count=200 -run TestTaskSlot_OperatorPanic ./internal/engine/... → 200 consecutive passes (was ~1 fail in 50 under -race). Full repo `go test -race ./...` green. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * docs(trds): WIP-24 fix mermaid sequence-diagram parse errors Two issues in the original diagrams: - <br/> inside note text is fragile in sequenceDiagram (works in flowchart, breaks here when followed by special chars). - A note line beginning with "-race" was parsed as an arrow start because Mermaid sees "-" after ":" and tries to consume an arrow token, then fails when the next line continues with "Writer-->>". Rewrite all three diagrams without <br/> and without lines starting with a dash. Same content, just inlined or split into separate Note lines. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * perf+feat: WIP-25 O(1) dup-check + job lifecycle metrics WIP-25: SubmitJob's duplicate-name check was an O(N) linear scan over c.jobs under c.mu.Lock(). With Pebble fsync no longer the bottleneck (WIP-23.1+2), the scan dominated the submit path once jobs accumulated. At 500 RPS load with ~70k CREATED jobs queued behind 4 worker slots: HTTP submit p99: 50ms → 4.7s (~95x spike), goroutines piled to 443. Pebble actually got faster during the spike (write_batch p99 49ms → 9ms via concurrent commit batching). GC, CPU, RSS all fine. The O(N*30ns) per-submit scan × 500 concurrent submits ≈ 1s of cumulative lock-queue depth. Fix: c.activeJobNames map[string]string (name -> jobID, non-terminal jobs only). SubmitJob dup check is now O(1). Maintained at three sites: - SubmitJob: insert on reservation; delete on persist failure rollback - transitionJob: delete on transition to terminal status (frees name) - recover(): rebuild from persisted non-terminal jobs Pause/resume preserves the reservation (JobPaused is non-terminal), matching the prior !j.Status.IsTerminal() filter semantic. Job lifecycle metrics (foundation for spotting WIP-25 in the first place): - wire.coordinator.job.duration histogram (terminal_status label), bucket boundaries 10ms..1h via a name-targeted view in otel.go so it overrides the catch-all latency view. - wire.coordinator.jobs.by_status observable gauge populated each scrape from c.jobStateCounts(). Bounded enum cardinality. - New "Coordinator jobs" row in the Grafana dashboard (active by status timeseries, p50/p99 by terminal_status, stacked completion rate bars). Tests: go test -race -count=20 -run "TestSubmitJob|TestRecovery" ./internal/coordinator/... is green; full coordinator suite is green. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
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.
No description provided.