feat: address WIP 26 improvments - #178
Draft
tarungka wants to merge 12 commits into
Draft
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>
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>
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>
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>
Catalog of optimization candidates across coordinator, engine, RPC,
transport, and SDK. Lands the low-risk and high-impact verified
findings in the same commit; the rest stay as candidates that need a
benchmark or production trace before promoting to their own WIP.
Implemented:
- C1: in-memory assignments cache; allTasksInStatus and CancelJob
no longer pay a Pebble Get + DecodeMsgPack per UpdateTaskStatus.
- C2: maintained jobStatusCounts; by-status gauge scrape is
O(distinct statuses), not O(N).
- C5: jobsByStatus secondary index; ListJobs(statusFilter) is
O(matched), not O(N).
- C3, C7: preallocated slices in flushHeartbeats and ListSavepoints.
- C10: validTransitions as map[from]map[to]struct{}.
- E1: processEvent fast path — no per-event []Event{event} alloc
unless a FlatMap fans out to >1.
- R1: ReadFrame allocates the body once and slices the payload out;
no more pool round-trip + copy.
- R9: lastWatermarks initialised in NewFrameStream, no lock-held
lazy alloc.
- S11: Worker.activeTasks atomic.Int32 mirrors len(w.tasks);
buildHeartbeatRequest reads the atomic.
Tests + race detector pass on all affected packages. The only red
test is TestCheckpointCoordinator_TimeoutAbort, a pre-existing flake
on master (verified under git stash).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
|
tarungka
changed the title
Analysis CompleteGenerated ECC bundle from 11 commits | Confidence: 55% View Pull Request #179Repository Profile
Changed Files (12)
Top hotspots
Top directories
Likely Future Issues (2)
Suggested Follow-up Work (2)
Copy-ready bodies test: add regression coverage for internal/coordinator/coordinator.go + internal/coordinator/job_manager.go ## Summary
- Add regression coverage for the recently touched code paths before more changes stack on top.
## Why
- Backfill regression coverage before another change set lands on the touched code paths.
## Touched paths
- `internal/coordinator/coordinator.go`
- `internal/coordinator/job_manager.go`
## Validation
- Add or extend focused tests that exercise the touched paths.
- Run the affected test suite and verify the new coverage closes the gap.test: add async coverage for internal/coordinator/scheduler.go + internal/worker/worker.go ## Summary
- Add async reliability coverage for the recently changed queue, worker, cron, or webhook surface.
## Why
- Backfill async reliability coverage before another queue, worker, or webhook change lands on the touched surface.
## Touched paths
- `internal/coordinator/scheduler.go`
- `internal/worker/worker.go`
## Validation
- Add or extend integration / e2e coverage for the changed queue, worker, cron, or webhook behavior.
- Exercise retries, idempotency, failure handling, or equivalent async boundary cases.Detected Workflows (3)
Generated Instincts (22)
After merging, import with: Files
|
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
🔒 Security Scan ResultsFound 27 issues (🔴 19 high · 🟠 8 medium · 🟡 0 low). Findings
…and 2 more. Full SARIF report on the repo's Security tab. |
CI fixes for PR #178: - gofmt: realign struct literal in coordinator.New and var block in operator_chain.processEvent. - gosec: annotate the two int → int32 conversions in worker.handleDeployTask / runTask with #nosec G115. len(w.tasks) is bounded by w.cfg.TaskSlots (a small config int), so the conversion cannot realistically approach 2^31. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
🔒 Security Scan ResultsFound 25 issues (🔴 17 high · 🟠 8 medium · 🟡 0 low). Findings
|
Owner
Author
|
Pausing this PR as the performance is actually more worse compared to the base branch. |
tarungka
marked this pull request as draft
May 12, 2026 12:48
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
2 participants
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.