feat: Plan 8 PR A — Go edge SSE client + atomic snapshot - #18
Conversation
Plan 8 PR A: immutable snapshot + apply, disk persist with dir fsync, control-plane client, SSE loop (shutdown/429 recovery), and proxy wiring to synced keys/scores/policy. Co-authored-by: Cursor <cursoragent@cursor.com>
|
Bugbot is not enabled for your account, so this pull request was not reviewed. Enable Bugbot in the Cursor dashboard to get automatic reviews on future PRs. |
|
Warning Review limit reachedYou’ve reached a temporary PR review limit under our Fair Usage Limits Policy. Next review available in: 58 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: ASSERTIVE Plan: Pro Run ID: 📒 Files selected for processing (5)
WalkthroughThe edge verifier now supports control-plane snapshot synchronization through in-memory state, disk persistence, HTTP snapshot fetching, reconnecting SSE events, policy-aware proxy decisions, and graceful signal-triggered shutdown. ChangesEdge-verifier synchronization
Estimated code review effort: 5 (Critical) | ~90 minutes Sequence Diagram(s)sequenceDiagram
participant EdgeVerifierProxy
participant SyncRunner
participant ControlPlaneClient
participant SnapshotStore
SyncRunner->>ControlPlaneClient: Fetch initial snapshot
ControlPlaneClient-->>SyncRunner: Return snapshot data
SyncRunner->>SnapshotStore: Apply snapshot
SyncRunner->>ControlPlaneClient: Open SSE session with cursor
ControlPlaneClient-->>SyncRunner: Deliver sync event
SyncRunner->>SnapshotStore: Apply event and update cursor
EdgeVerifierProxy->>SnapshotStore: Read policy, score, or key
SnapshotStore-->>EdgeVerifierProxy: Return snapshot-backed decision data
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Comment |
Co-authored-by: Cursor <cursoragent@cursor.com>
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
PR Summary by QodoGo edge: control-plane SSE sync with atomic snapshot + stale-mode gating
AI Description
Diagram
High-Level Assessment
Files changed (11)
|
messagesgoel-blip
left a comment
messagesgoel-blip
left a comment
There was a problem hiding this comment.
Review: PR #18 — Plan 8 PR A: Go edge SSE client + atomic snapshot
This is a substantial, well-structured PR. 1630 additions across 11 files implementing the Go-side edge sync client.
Architecture
| Component | Purpose |
|---|---|
snapshot.go |
Store (atomic.Pointer[Snapshot]), ScoreEntry, KeyEntry, Policy, lookup helpers with key validity windows |
apply.go |
ApplySnapshot, ApplyEvent (idempotent), AdvanceCursor, CloneSnapshot |
cpclient.go |
ControlPlaneClient.FetchSnapshot with gzip support |
disk.go |
Crash-safe save (tmp→fsync→rename→dir fsync) and load with schema versioning |
sse.go |
SyncRunner: bootstrap, reconnect loop with backoff, SSE protocol parser, 429/410 recovery |
policy.go |
EvaluateMode (stale→503, expired→X-Verilink-Mode), AllowByScore, RequireSignaturesFromPolicy |
proxy.go |
Modified to use snapshot store for key/score/policy lookups |
main.go |
Wiring: flags, bootstrap, sync runner goroutine, signal handling |
What is done well
- Atomic snapshot swap —
Storeusesatomic.Pointerfor lock-free concurrent reads from the proxy - Crash-safe disk persistence — tmp → fsync → rename → dir fsync pattern
- Idempotent event apply —
syncVersion <= hwskips already-applied events - SSE protocol parser — correctly handles event/id/data/retry/comment lines and blank-line dispatch
- 429/410 recovery —
runSessionfetches a fresh snapshot on those status codes and reconnects with the new cursor event: shutdownhandling — returns a sentinel error triggering reconnect- Exponential backoff — 5s default, 30s max
- Key validity windows —
LookupKeychecksValidFrom/ValidUntil - Tests —
snapshot_test.go(164 lines) andsse_test.go(143 lines) covering atomic swap, idempotent apply, disk round-trip, evaluate mode, key validity, shutdown reconnect, and 429 recovery with snapshot fetch
Issues
1. AdvanceCursor clones the entire snapshot on every cursor event (Performance — high impact)
func AdvanceCursor(store *Store, highWater int64) {
next := CloneSnapshot(cur) // deep-copies ALL scores + keys
next.HighWaterVersion = highWater
store.Swap(next)
}Cursor events fire every poll cycle (every 5s). On a tenant with 100K scores, this means cloning 100K map entries every 5 seconds — even when no data changed. This is O(n) per poll cycle for the entire lifetime of the SSE connection.
Fix: Store HighWaterVersion as a separate atomic.Int64 on Store instead of inside Snapshot. Then AdvanceCursor just needs atomic.Store instead of clone+swap. The Snapshot HW field becomes a bootstrap-only value.
2. No SetSyncReachable(false) when SSE disconnects (Correctness — medium impact)
The proxys EvaluateMode checks syncReachable via p.syncReachable.Load(), but nothing ever sets it to false. The SyncRunner.Run loop has no reference to the proxy. So syncReachable stays true forever, and stale detection is entirely reactive (waits for BytesAge to exceed policy threshold).
Fix: Pass a SetSyncReachable func(bool) into SyncRunner (or ControlPlaneClient), called when an SSE session starts (true) and when it fails (false). The proxy calls SetSyncReachable to trigger proactive stale detection.
3. parseSnapshotJSON fallback swallows original error context (Minor)
When both the wrapped and bare unmarshal fail, the function returns the first error, not the second. If the bare format is close but has a typo, the error message may mislead about what went wrong.
Minor observations
main.gousessyscallfor signal handling — fine, butos.Signalandos/signalare sufficient for SIGINT/SIGTERMhttp.Client{Timeout: 0}is used for the SSE client — correct (no timeout for long-lived streams)- The 30s bootstrap timeout is a good default for the initial snapshot fetch
envOrhelper is defined at the bottom ofmain.go— a small utility, fine to keep local
Verdict
Changes requested — issue #1 (AdvanceCursor O(n) clone on every poll cycle) is a significant performance concern at scale. Issue #2 (SetSyncReachable never called) is a correctness gap for proactive stale detection. Everything else is high quality and well-tested.
Code Review by Qodo
Context used✅ Compliance rules (platform):
18 rules 1.
|
![qodo-code-review[bot]](https://avatars.githubusercontent.com/in/484649?s=60&v=4){kind=small}
Store high-water as atomic.Int64 (no cursor clone), wire OnReachable, nil-safe policy helpers, clamp retry, reject bad event IDs / HW, cap snapshot reads, validate key.upsert IDs, and clear golangci-lint. Co-authored-by: Cursor <cursoragent@cursor.com>
|
Addressed CI lint + review findings:
@coderabbitai review |
|
✅ Action performedReview finished.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 12
🤖 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 `@cmd/edge-verifier/main.go`:
- Line 35: Remove the api-key flag registration from the command-line
configuration, ensuring the control-plane API key is sourced only through the
existing VERILINK_API_KEY environment variable path in the edge verifier startup
flow.
- Around line 86-88: Bound control-plane operations without adding an overall
SSE timeout: in cmd/edge-verifier/main.go lines 86-88, clone the default HTTP
transport and configure a finite ResponseHeaderTimeout for connection
establishment; in internal/edgeverifier/sse.go lines 138-148, apply that header
deadline while keeping the established response body long-lived; in
internal/edgeverifier/sse.go lines 171-177, derive and use a finite timeout
context for each recovery snapshot fetch.
In `@internal/edgeverifier/apply.go`:
- Around line 176-190: Validate p.BelowThresholdAction and p.UnsignedAction
after applying their defaults and before returning the policy, allowing only
allow or deny for the former and passthrough or deny for the latter. Return an
appropriate error for unsupported values so they cannot be published into the
live enforcement policy.
- Around line 62-69: Validate p.PrincipalID in the score.delete handling before
deleting from next.Scores or acknowledging the event; return an error for an
empty identifier so the high-water mark is not advanced. Keep valid deletions
unchanged.
In `@internal/edgeverifier/cpclient.go`:
- Around line 154-190: Add one shared full-snapshot validator and invoke it
before publishing or inserting entries: internal/edgeverifier/cpclient.go lines
154-190 must reject empty or duplicate principal/key IDs and invalid key
validity windows, while internal/edgeverifier/disk.go lines 113-153 must use the
same validation, reject negative high-water values, and normalize policy
defaults consistently before returning. Ensure both ingestion paths use the
validator rather than maintaining separate checks.
In `@internal/edgeverifier/disk.go`:
- Around line 103-110: Update LoadSnapshot to read through an io.LimitedReader
capped by MaxSnapshotBytes plus one byte, rather than using os.ReadFile
directly; reject input containing the extra byte before calling json.Unmarshal,
while preserving the existing error propagation and snapshot parsing behavior.
In `@internal/edgeverifier/policy.go`:
- Around line 17-41: Update EvaluateMode and Store.BytesAge so the supplied now
timestamp is used for staleness calculation, changing BytesAge to accept now and
passing it from EvaluateMode; remove the unused _ = now statement while
preserving existing policy and mode behavior.
In `@internal/edgeverifier/sse_test.go`:
- Around line 15-18: Update the httptest server handler in the SSE test to count
sessions only after confirming the request path is /v1/sync/events, using an
atomic counter to avoid races with the polling goroutine. Use the per-request
atomic count for the first-session shutdown branch, and keep the assertion
validating at least two SSE sessions; verify with go test ./....
In `@internal/edgeverifier/sse.go`:
- Around line 224-235: Add an idle watchdog around the SSE read loop in the
relevant runner, using the heartbeat freshness deadline to close the response
body when no data arrives; ensure the blocked ReadString unblocks and returns
through the existing reconnect/error path, and stop the watchdog when the stream
exits.
- Around line 228-264: Update the SSE parsing loop around ReadString and
dataLines to enforce explicit maximum line and assembled-event sizes. Reject or
return an appropriate error when a line exceeds the configured maximum, and
check the aggregate size before appending each data: value so oversized events
cannot grow memory unbounded; preserve normal flush, retry, and event parsing
behavior.
- Around line 104-130: Preserve exponential backoff in the reconnect loop around
runSession by only updating backoffMs when the server provides a positive retry
value, rather than treating the session functions’ default retryMs of 5000 as an
override on every failure. Ensure the same behavior is applied to the
corresponding retry-handling blocks at the other referenced session paths, while
retaining the existing backoffMs doubling when no retry field is sent.
- Around line 214-220: Update the event handling around ApplyEvent so an error
applying a known event terminates the session instead of returning nil and
continuing. Preserve the existing error log, but return the appropriate non-nil
session-processing error from this branch so later events cannot advance the
high-water mark past the rejected event; keep successful events on the existing
persist-and-return path.
🪄 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: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 78b61405-5567-41ab-a090-a0e63ece6ad3
📒 Files selected for processing (11)
cmd/edge-verifier/main.godocs/superpowers/plans/HANDOVER.mdinternal/edgeverifier/apply.gointernal/edgeverifier/cpclient.gointernal/edgeverifier/disk.gointernal/edgeverifier/policy.gointernal/edgeverifier/proxy.gointernal/edgeverifier/snapshot.gointernal/edgeverifier/snapshot_test.gointernal/edgeverifier/sse.gointernal/edgeverifier/sse_test.go
Env-only API key, snapshot validation, SSE idle/size bounds, preserve backoff without retry:, and fail sessions on apply errors. Co-authored-by: Cursor <cursoragent@cursor.com>
|
Addressed all 12 CodeRabbit findings in
@coderabbitai review |
|
✅ Action performedReview finished.
|
CI Gate failed with complexity 26 on both functions after the validation/guard additions; extract event and wire helpers. Co-authored-by: Cursor <cursoragent@cursor.com>
|
@coderabbitai review — also fixed gocyclo Gate failure in follow-up commit. |
|
✅ Action performedReview finished.
|
Co-authored-by: Cursor <cursoragent@cursor.com>
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
internal/edgeverifier/sse.go (2)
81-100: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Bootstrap's snapshot fetch has no timeout, unlikerecoverSnapshot.
recoverSnapshot(Line 177) wrapsFetchSnapshotincontext.WithTimeout(ctx, snapshotFetchTO), butBootstrap's equivalent call at Line 90 uses the rawctxdirectly. If the control plane hangs during startup (no TCP reset, just silence) andctxhas no deadline,Bootstrap— and thereforeRun— can block indefinitely with no recovery path, defeating the purpose of thesnapshotFetchTOguard established elsewhere in this same file.🐛 Proposed fix
- snap, err := r.Client.FetchSnapshot(ctx) + fetchCtx, cancel := context.WithTimeout(ctx, snapshotFetchTO) + defer cancel() + snap, err := r.Client.FetchSnapshot(fetchCtx) if err != nil { return fmt.Errorf("fetch snapshot: %w", err) }🤖 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 `@internal/edgeverifier/sse.go` around lines 81 - 100, Update SyncRunner.Bootstrap to wrap the FetchSnapshot call with the existing snapshotFetchTO timeout, matching recoverSnapshot’s behavior, and pass the timed context to FetchSnapshot so startup cannot block indefinitely when the parent context has no deadline.
257-268: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winProcess the final SSE line before returning on
io.EOF
readSSELinecan return a non-emptylinetogether withio.EOF, but this loop exits before dispatching it. That drops a final unterminateddata:line and can lose the last SSE event. Handle the buffered line first, then flush and return on EOF.🤖 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 `@internal/edgeverifier/sse.go` around lines 257 - 268, Update the SSE read loop around readSSELine to process any non-empty line returned alongside io.EOF before flushing and returning. Preserve the existing EOF flush and return behavior afterward, ensuring the final unterminated data line is dispatched.internal/edgeverifier/sse_test.go (1)
82-144: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winUse an atomic counter for the snapshot request count
nis incremented in the httptest handler and read from the test goroutine, so this test still has an unsynchronized cross-goroutine access. Switch it toatomic.Int32or guard it with a mutex to avoid-racefailures.🤖 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 `@internal/edgeverifier/sse_test.go` around lines 82 - 144, The snapshot request counter n is accessed concurrently without synchronization. Replace n with an atomic.Int32 (or protect it with a mutex), update its increment in the httptest handler and load in the final t.Fatalf message, and preserve the existing request-count behavior.Source: Path instructions
🤖 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 `@internal/edgeverifier/apply.go`:
- Around line 108-152: Update applyKeyUpsert to validate that a non-nil
ValidUntil is strictly after ValidFrom before inserting the KeyEntry, matching
validateSnapshot’s full-snapshot behavior. Return a key.upsert validation error
and avoid mutating next.Keys when the validity window is inverted or
zero-length.
In `@internal/edgeverifier/cpclient.go`:
- Around line 201-231: Unify public-key decoding by moving the raw-base64
fallback into DecodePublicKeyRaw, then remove the one-off fallback from
wireKeyToEntry so applyKeyUpsert and LoadSnapshot share identical behavior. In
wireKeyToEntry, return an explicit error when the decoded key length is not 32
bytes instead of wrapping a potentially nil error.
---
Outside diff comments:
In `@internal/edgeverifier/sse_test.go`:
- Around line 82-144: The snapshot request counter n is accessed concurrently
without synchronization. Replace n with an atomic.Int32 (or protect it with a
mutex), update its increment in the httptest handler and load in the final
t.Fatalf message, and preserve the existing request-count behavior.
In `@internal/edgeverifier/sse.go`:
- Around line 81-100: Update SyncRunner.Bootstrap to wrap the FetchSnapshot call
with the existing snapshotFetchTO timeout, matching recoverSnapshot’s behavior,
and pass the timed context to FetchSnapshot so startup cannot block indefinitely
when the parent context has no deadline.
- Around line 257-268: Update the SSE read loop around readSSELine to process
any non-empty line returned alongside io.EOF before flushing and returning.
Preserve the existing EOF flush and return behavior afterward, ensuring the
final unterminated data line is dispatched.
🪄 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: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 68fa75e0-b524-4fc8-ba32-b31661d40795
📒 Files selected for processing (8)
cmd/edge-verifier/main.gointernal/edgeverifier/apply.gointernal/edgeverifier/cpclient.gointernal/edgeverifier/disk.gointernal/edgeverifier/policy.gointernal/edgeverifier/sse.gointernal/edgeverifier/sse_test.gointernal/edgeverifier/validate.go
Align key.upsert validity and key decoding across ingest paths, add bootstrap snapshot timeout, handle trailing SSE EOF lines, and make the 429 recovery counter race-safe. Co-authored-by: Cursor <cursoragent@cursor.com>
|
Addressed the latest CodeRabbit CHANGES_REQUESTED findings:
Also extracted SSE frame parsing to keep gocyclo under the Gate limit. @coderabbitai review |
|
✅ Action performedReview finished.
|
messagesgoel-blip
left a comment
messagesgoel-blip
left a comment
There was a problem hiding this comment.
Re-review: PR #18
The 5 new commits thoroughly address all concerns from my previous review.
What was fixed
| Previous issue | Fix |
|---|---|
| AdvanceCursor clones entire snapshot every poll cycle | Store now has a separate atomic.Int64 for high-water. advanceHighWater uses CompareAndSwap — no clone. |
| SetSyncReachable never called | SyncRunner has OnReachable callback. Wired via WithOnReachable(proxy.SetSyncReachable). |
| API key in argv | -api-key flag removed. VERILINK_API_KEY is env-only. |
Additional hardening added
- SSE idle watchdog (90s timeout)
- Line/event byte limits (256KiB / 1MiB)
ResponseHeaderTimeouton HTTP transport- 30s timeout constant for all snapshot fetches
- RawStdEncoding fallback for key decoding
- Cyclomatic complexity split for lint compliance
Verdict
All concerns resolved, plus meaningful hardening beyond what was requested. Ready to merge.
1 participant
Summary
internal/edgeverifiersnapshot store (atomic swap), event apply (idempotent), disk persist (tmp → fsync → rename → dir fsync)retryms,shutdownreconnect,429/410snapshot recovery503; keep demo MockTrustStore path when sync offVERILINK_CONTROL_PLANE_URL,VERILINK_API_KEY,VERILINK_SNAPSHOT_PATHTest plan
go test ./internal/edgeverifier/... ./cmd/edge-verifier/...go vet ./internal/edgeverifier/... ./cmd/edge-verifier/...@coderabbitai review
Summary by CodeRabbit