feat: Plan 8 PR C — decision batch ingest + HTTP flush - #21
Conversation
Add POST /v1/decisions/batch with idempotent (edge_node, batch_id) delivery into decision_batches/aggregates/samples, wire the edge FlushWorker to ControlPlaneClient when sync is enabled, and refresh handover for the dashboard next step. Co-authored-by: Cursor <cursoragent@cursor.com>
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. |
|
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. |
WalkthroughThe PR adds authenticated control-plane decision batch ingestion, canonical validation and hashing, aggregate and sample persistence, idempotency handling, and HTTP flushing from the edge verifier with retry backoff. ChangesDecision ingestion flow
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant EdgeVerifier
participant ControlPlaneClient
participant DecisionsRoute
participant DecisionIngestService
participant PostgreSQL
EdgeVerifier->>ControlPlaneClient: Build and flush canonical decision batch
ControlPlaneClient->>DecisionsRoute: POST /v1/decisions/batch
DecisionsRoute->>DecisionIngestService: Authenticate and ingest batch
DecisionIngestService->>PostgreSQL: Write batch, aggregates, and samples
PostgreSQL-->>DecisionIngestService: Return commit or duplicate outcome
DecisionIngestService-->>DecisionsRoute: Return ingestion result
DecisionsRoute-->>ControlPlaneClient: Return HTTP response
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Comment |
PR Summary by QodoDecision batch ingest endpoint + edge HTTP flush transport
AI Description
Diagram
High-Level Assessment
Files changed (9)
|
Code Review by Qodo
Context used✅ Compliance rules (platform):
18 rules 1.
|
![qodo-code-review[bot]](https://avatars.github.com/in/484649?s=60&v=4){kind=small}
Verify payload_hash/batch_id, bulk UNNEST upserts, decided_at sampling, RFC3339 validation, flush backoff, and canonical decisionWire hashing. Co-authored-by: Cursor <cursoragent@cursor.com>
![coderabbitai[bot]](https://avatars.github.com/in/347564?s=60&v=4){kind=small}
|
Addressed open Qodo threads + the pasted human review in
@coderabbitai review |
|
✅ Action performedReview finished.
|
golangci gosec G404 rejects math/rand for the retry jitter path. Co-authored-by: Cursor <cursoragent@cursor.com>
Use crypto/rand.Int with math/big instead of uint64→int64 casting. Co-authored-by: Cursor <cursoragent@cursor.com>
Document that FlushBatch hash scope excludes the wire envelope so maintainers do not accidentally widen the digest. Co-authored-by: Cursor <cursoragent@cursor.com>
|
Applied the remaining review suggestion in the latest commit: clarified on |
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 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 `@control-plane/src/__tests__/integration/decision-ingest.test.ts`:
- Around line 61-77: Extend the decision ingest integration tests around the
existing payload_hash mismatch case to cover the remaining assertBatchShape
branches: submit a buildBatch result with an overridden batch_id and expect 400,
submit more than 500 decisions and expect 400, and submit inconsistent
first_wal_seq/last_wal_seq values and expect 400. Reuse the existing harness,
authHeaders, buildBatch, and response assertion patterns.
In `@control-plane/src/domains/decision/decisionIngestService.ts`:
- Around line 232-277: Update the transaction cleanup around the catch/finally
blocks to track whether ROLLBACK fails, and pass that state to
client.release(true) when it does. Preserve normal release behavior for
successful rollback or non-transaction failures, while ensuring rollback errors
are no longer silently returned as reusable connections.
In `@internal/edgeverifier/cpclient.go`:
- Around line 98-104: Sanitize the response text in the non-OK branch of the
decision-batch request before passing it to truncate and fmt.Errorf. Strip or
escape newline, carriage-return, and other control characters from errBody so
the error returned by the surrounding request function remains safe when logged
by flush handling, while preserving the existing status and truncation behavior.
In `@internal/edgeverifier/flush.go`:
- Around line 139-177: Re-clamp the retry delay after applying jitter in the
FlushOnce error path, using the existing maxBackoff limit so backoff never
exceeds 30 seconds. Keep the current exponential backoff and jitter behavior,
but ensure the final assigned backoff value is bounded by maxBackoff.
🪄 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: 38a2963b-52a9-4826-aa31-b21983fcb49c
📒 Files selected for processing (9)
cmd/edge-verifier/main.gocontrol-plane/src/__tests__/integration/decision-ingest.test.tscontrol-plane/src/app.tscontrol-plane/src/domains/decision/decisionIngestService.tscontrol-plane/src/domains/decision/decisionSample.tscontrol-plane/src/routes/decisions.tsdocs/superpowers/plans/HANDOVER.mdinternal/edgeverifier/cpclient.gointernal/edgeverifier/flush.go
| it('rejects payload_hash that does not match decisions', async () => { | ||
| const decidedAt = new Date().toISOString(); | ||
| const batch = buildBatch([ | ||
| { | ||
| wal_seq: 1, | ||
| fingerprint: 'fp-a', | ||
| action: 'deny', | ||
| decided_at: decidedAt, | ||
| }, | ||
| ]); | ||
| const resp = await fetch(`${harness.url}/v1/decisions/batch`, { | ||
| method: 'POST', | ||
| headers: { ...authHeaders(apiKey), 'Content-Type': 'application/json' }, | ||
| body: JSON.stringify({ ...batch, payload_hash: '0'.repeat(64) }), | ||
| }); | ||
| assert.equal(resp.status, 400); | ||
| }); |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Add coverage for the remaining assertBatchShape validation branches.
The current tests cover payload_hash-vs-decisions mismatch, idempotent redelivery, stored-hash conflict, and non-RFC3339 decided_at. They do not cover the batch_id does not match payload_hash branch (send a buildBatch result with batch_id overridden instead of payload_hash), the decisions.length > 500 limit, or the first_wal_seq/last_wal_seq mismatch check. These are new validations called out in the PR comments summary.
🤖 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 `@control-plane/src/__tests__/integration/decision-ingest.test.ts` around lines
61 - 77, Extend the decision ingest integration tests around the existing
payload_hash mismatch case to cover the remaining assertBatchShape branches:
submit a buildBatch result with an overridden batch_id and expect 400, submit
more than 500 decisions and expect 400, and submit inconsistent
first_wal_seq/last_wal_seq values and expect 400. Reuse the existing harness,
authHeaders, buildBatch, and response assertion patterns.
| const client = await pool.connect(); | ||
| let began = false; | ||
| try { | ||
| await client.query('BEGIN'); | ||
| began = true; | ||
| await client.query( | ||
| `INSERT INTO decision_batches | ||
| (edge_node_id, batch_id, tenant_id, first_wal_seq, last_wal_seq, payload_hash) | ||
| VALUES ($1, $2, $3, $4, $5, $6)`, | ||
| [ | ||
| edgeNodeId, | ||
| opts.body.batch_id, | ||
| opts.tenantId, | ||
| opts.body.first_wal_seq, | ||
| opts.body.last_wal_seq, | ||
| payloadHash, | ||
| ] | ||
| ); | ||
| await insertAggregates(client, opts.tenantId, edgeNodeId, [...aggMap.values()]); | ||
| await insertSamples(client, opts.tenantId, edgeNodeId, samples); | ||
| await client.query('COMMIT'); | ||
| } catch (err) { | ||
| if (began) { | ||
| try { | ||
| await client.query('ROLLBACK'); | ||
| } catch { | ||
| // ignore rollback errors; connection will be discarded by release | ||
| } | ||
| } | ||
| if ((err as { code?: string }).code === '23505') { | ||
| const again = await pool.query( | ||
| `SELECT payload_hash FROM decision_batches WHERE edge_node_id = $1 AND batch_id = $2`, | ||
| [edgeNodeId, opts.body.batch_id] | ||
| ); | ||
| if ( | ||
| again.rows.length > 0 && | ||
| String(again.rows[0].payload_hash).toLowerCase() === payloadHash | ||
| ) { | ||
| return { duplicate: true, edge_node_id: edgeNodeId, accepted: 0, sampled: 0 }; | ||
| } | ||
| throw new AppError(CODES.CONFLICT, 'batch_id reused with different payload_hash'); | ||
| } | ||
| throw err; | ||
| } finally { | ||
| client.release(); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
cd /srv/storage/repo/VeriLink && find . -name "package.json" -type f | head -5Repository: Numeracode/verilink
Length of output: 2026
🏁 Script executed:
cd /srv/storage/repo/VeriLink && cat package.json | grep -A2 -B2 '"pg"'Repository: Numeracode/verilink
Length of output: 2026
🏁 Script executed:
cd /srv/storage/repo/VeriLink && rg "client\.release\(" --type=typescript --type=javascript -B2 -A2 | head -80Repository: Numeracode/verilink
Length of output: 2026
🌐 Web query:
node-postgres pg client.release(err) error handling discard connection
💡 Result:
In node-postgres, the client.release(err) method is the standard way to return an acquired client to the pool [1][2]. The err argument serves as a flag to control whether the client should be returned to the pool for reuse or destroyed [2]. When you call client.release(err): 1. If err is truthy (e.g., an error object or true), the pool is instructed to discard the client (close the connection) instead of returning it to the pool for future use [2]. This is useful when you have encountered a fatal error on the connection and want to ensure the pool replaces it with a fresh, clean client [2]. 2. If no argument (or a falsy value) is passed, the client is returned to the pool as an idle, available connection [2]. Important Considerations: - You must always release the client after it is acquired via pool.connect to avoid connection leaks [1][3][2]. - To handle errors correctly without leaks, ensure your code paths always call release [3][2]. A common pattern using try/finally ensures the client is released regardless of whether the query succeeded or failed: javascript const client = await pool.connect(); try { await client.query('SELECT...'); } catch (err) { // Pass error to release to discard the potentially corrupted connection client.release(err); throw err; // Re-throw if necessary } finally { // If the client was already released in catch, this is a no-op // (though it is cleaner to ensure it only happens once) client.release(); } - Be aware that calling release twice can cause issues depending on your implementation; it is best practice to ensure your logic calls release exactly once [4]. - Do not use client.end on a client acquired from a pool, as it will close the connection and potentially hang your process or prevent it from returning to the pool correctly [5]. Only use client.release for pooled clients [5].
Citations:
- 1: Docs: document client.release() brianc/node-postgres#1302
- 2: https://github.com/brianc/node-postgres-docs/blob/master/content/api/1-pool.mdx
- 3: https://node-postgres.com/features/pooling
- 4: feat: evict connections with open transactions on pool release brianc/node-postgres#3597
- 5: client.release and pitfall with client.end are not documented in pg.Client’s API brianc/node-postgres#2101
🌐 Web query:
node-postgres PoolClient release() method semantics aborted transaction
💡 Result:
In node-postgres, the poolClient.release method does not automatically rollback or commit open transactions [1]. When you call release, the client is returned to the pool in whatever state it was left in [1]. If you leave a transaction open (e.g., you issued BEGIN but did not issue a COMMIT or ROLLBACK) and then release the client, the transaction remains open [1]. Because the client is returned to the pool, the next piece of code that acquires this same client will inherit the existing, uncommitted transaction state [1]. This can lead to unexpected behavior, such as subsequent queries accidentally becoming part of a previous transaction or failing due to state conflicts [2][1]. To safely handle transactions: 1. Always implement an explicit ROLLBACK in your error handling (catch block) [3][1]. 2. Always ensure the client is returned to the pool in the finally block, regardless of success or failure [3][1]. 3. If an error occurs during your ROLLBACK attempt, you should treat the client as broken and pass a truthy value to release(true) (or destroy it) to ensure it is not reused by the pool [4][3][5]. Note: Recent developments in node-postgres (as of 2026) have introduced an optional pool configuration setting evictOnOpenTransaction. If enabled, this allows the pool to automatically evict (close) connections released while they have an active, uncommitted transaction [6]. However, this is not the default behavior and must be explicitly configured [6].
Citations:
- 1: Record not able to insert properly using pool.query brianc/node-postgres#3286
- 2: https://node-postgres.com/apis/pool
- 3: Helper functions brianc/node-postgres#2107
- 4: "Client was closed and is not queryable" when using client from pool in transactions brianc/node-postgres#2512
- 5: https://github.com/brianc/node-postgres-docs/blob/master/content/api/1-pool.mdx
- 6: feat: evict connections with open transactions on pool release brianc/node-postgres#3597
Pass the rollback error to client.release() to discard poisoned connections.
If ROLLBACK throws in the catch block, the error is caught and discarded, then client.release() is called with no argument. When ROLLBACK fails, the database session can remain in an aborted-transaction state. Calling release() without an argument returns this poisoned connection to the pool. The next code that acquires this connection inherits the aborted-transaction state and fails on every query, cascading the failure to unrelated requests.
Track the rollback failure with a boolean flag and pass it to release(true). The pool will then discard the connection instead of reusing it.
🔧 Proposed fix
const client = await pool.connect();
let began = false;
+ let rollbackFailed = false;
try {
await client.query('BEGIN');
began = true;
...
await client.query('COMMIT');
} catch (err) {
if (began) {
try {
await client.query('ROLLBACK');
} catch {
- // ignore rollback errors; connection will be discarded by release
+ rollbackFailed = true;
}
}
...
throw err;
} finally {
- client.release();
+ client.release(rollbackFailed);
}📝 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.
| const client = await pool.connect(); | |
| let began = false; | |
| try { | |
| await client.query('BEGIN'); | |
| began = true; | |
| await client.query( | |
| `INSERT INTO decision_batches | |
| (edge_node_id, batch_id, tenant_id, first_wal_seq, last_wal_seq, payload_hash) | |
| VALUES ($1, $2, $3, $4, $5, $6)`, | |
| [ | |
| edgeNodeId, | |
| opts.body.batch_id, | |
| opts.tenantId, | |
| opts.body.first_wal_seq, | |
| opts.body.last_wal_seq, | |
| payloadHash, | |
| ] | |
| ); | |
| await insertAggregates(client, opts.tenantId, edgeNodeId, [...aggMap.values()]); | |
| await insertSamples(client, opts.tenantId, edgeNodeId, samples); | |
| await client.query('COMMIT'); | |
| } catch (err) { | |
| if (began) { | |
| try { | |
| await client.query('ROLLBACK'); | |
| } catch { | |
| // ignore rollback errors; connection will be discarded by release | |
| } | |
| } | |
| if ((err as { code?: string }).code === '23505') { | |
| const again = await pool.query( | |
| `SELECT payload_hash FROM decision_batches WHERE edge_node_id = $1 AND batch_id = $2`, | |
| [edgeNodeId, opts.body.batch_id] | |
| ); | |
| if ( | |
| again.rows.length > 0 && | |
| String(again.rows[0].payload_hash).toLowerCase() === payloadHash | |
| ) { | |
| return { duplicate: true, edge_node_id: edgeNodeId, accepted: 0, sampled: 0 }; | |
| } | |
| throw new AppError(CODES.CONFLICT, 'batch_id reused with different payload_hash'); | |
| } | |
| throw err; | |
| } finally { | |
| client.release(); | |
| } | |
| const client = await pool.connect(); | |
| let began = false; | |
| let rollbackFailed = false; | |
| try { | |
| await client.query('BEGIN'); | |
| began = true; | |
| await client.query( | |
| `INSERT INTO decision_batches | |
| (edge_node_id, batch_id, tenant_id, first_wal_seq, last_wal_seq, payload_hash) | |
| VALUES ($1, $2, $3, $4, $5, $6)`, | |
| [ | |
| edgeNodeId, | |
| opts.body.batch_id, | |
| opts.tenantId, | |
| opts.body.first_wal_seq, | |
| opts.body.last_wal_seq, | |
| payloadHash, | |
| ] | |
| ); | |
| await insertAggregates(client, opts.tenantId, edgeNodeId, [...aggMap.values()]); | |
| await insertSamples(client, opts.tenantId, edgeNodeId, samples); | |
| await client.query('COMMIT'); | |
| } catch (err) { | |
| if (began) { | |
| try { | |
| await client.query('ROLLBACK'); | |
| } catch { | |
| rollbackFailed = true; | |
| } | |
| } | |
| if ((err as { code?: string }).code === '23505') { | |
| const again = await pool.query( | |
| `SELECT payload_hash FROM decision_batches WHERE edge_node_id = $1 AND batch_id = $2`, | |
| [edgeNodeId, opts.body.batch_id] | |
| ); | |
| if ( | |
| again.rows.length > 0 && | |
| String(again.rows[0].payload_hash).toLowerCase() === payloadHash | |
| ) { | |
| return { duplicate: true, edge_node_id: edgeNodeId, accepted: 0, sampled: 0 }; | |
| } | |
| throw new AppError(CODES.CONFLICT, 'batch_id reused with different payload_hash'); | |
| } | |
| throw err; | |
| } finally { | |
| client.release(rollbackFailed); | |
| } |
🤖 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 `@control-plane/src/domains/decision/decisionIngestService.ts` around lines 232
- 277, Update the transaction cleanup around the catch/finally blocks to track
whether ROLLBACK fails, and pass that state to client.release(true) when it
does. Preserve normal release behavior for successful rollback or
non-transaction failures, while ensuring rollback errors are no longer silently
returned as reusable connections.
| if resp.StatusCode == http.StatusOK { | ||
| _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 64<<10)) | ||
| return nil | ||
| } | ||
| errBody, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) | ||
| return fmt.Errorf("decision batch: status %d: %s", resp.StatusCode, truncate(string(errBody), 200)) | ||
| } |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟡 Minor | ⚡ Quick win
Strip control characters from the response body before embedding it in the error/log chain.
errBody is raw content from the control-plane's HTTP response and is embedded verbatim (truncated to 200 bytes) into the returned error. That error later reaches w.logf("flush failed: %v", err) in internal/edgeverifier/flush.go (line 230), which writes it straight to the log via Printf — flagged by static analysis for log injection (CWE-117). If the response body contains newlines or control characters, an attacker who can influence that response (e.g. a compromised control-plane or a misbehaving intermediary) can forge or split log lines. Strip or escape control characters (at least \n/\r) from errBody before it becomes part of the error message.
🤖 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/cpclient.go` around lines 98 - 104, Sanitize the
response text in the non-OK branch of the decision-batch request before passing
it to truncate and fmt.Errorf. Strip or escape newline, carriage-return, and
other control characters from errBody so the error returned by the surrounding
request function remains safe when logged by flush handling, while preserving
the existing status and truncation behavior.
Source: Linters/SAST tools
| backoff := time.Duration(0) | ||
| for { | ||
| wait := interval | ||
| if backoff > 0 { | ||
| wait = backoff | ||
| } | ||
| timer := time.NewTimer(wait) | ||
| select { | ||
| case <-ctx.Done(): | ||
| if !timer.Stop() { | ||
| select { | ||
| case <-timer.C: | ||
| default: | ||
| } | ||
| } | ||
| flushCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) | ||
| _ = w.FlushAll(flushCtx) | ||
| cancel() | ||
| return | ||
| case <-ticker.C: | ||
| case <-timer.C: | ||
| flushCtx, cancel := context.WithTimeout(ctx, 5*time.Second) | ||
| _ = w.FlushOnce(flushCtx) | ||
| err := w.FlushOnce(flushCtx) | ||
| cancel() | ||
| if err != nil { | ||
| if backoff == 0 { | ||
| backoff = interval | ||
| } else { | ||
| backoff *= 2 | ||
| } | ||
| const maxBackoff = 30 * time.Second | ||
| if backoff > maxBackoff { | ||
| backoff = maxBackoff | ||
| } | ||
| // ±20% jitter | ||
| j := time.Duration(randInt63n(int64(backoff/5) + 1)) | ||
| backoff = backoff - backoff/10 + j | ||
| } else { | ||
| backoff = 0 | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Re-clamp backoff after applying jitter.
maxBackoff clamps backoff before the jitter adjustment (backoff - backoff/10 + j), but that adjustment can add back up to 10% on top of the already-clamped value. The next wait can then reach roughly 33 seconds, exceeding the documented 30-second cap stated in the comment above Run and in the PR description ("capped at 30 seconds"). Re-clamp after computing jitter.
🔧 Proposed fix
// ±20% jitter
j := time.Duration(randInt63n(int64(backoff/5) + 1))
backoff = backoff - backoff/10 + j
+ if backoff > maxBackoff {
+ backoff = maxBackoff
+ }
} else {
backoff = 0
}📝 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.
| backoff := time.Duration(0) | |
| for { | |
| wait := interval | |
| if backoff > 0 { | |
| wait = backoff | |
| } | |
| timer := time.NewTimer(wait) | |
| select { | |
| case <-ctx.Done(): | |
| if !timer.Stop() { | |
| select { | |
| case <-timer.C: | |
| default: | |
| } | |
| } | |
| flushCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) | |
| _ = w.FlushAll(flushCtx) | |
| cancel() | |
| return | |
| case <-ticker.C: | |
| case <-timer.C: | |
| flushCtx, cancel := context.WithTimeout(ctx, 5*time.Second) | |
| _ = w.FlushOnce(flushCtx) | |
| err := w.FlushOnce(flushCtx) | |
| cancel() | |
| if err != nil { | |
| if backoff == 0 { | |
| backoff = interval | |
| } else { | |
| backoff *= 2 | |
| } | |
| const maxBackoff = 30 * time.Second | |
| if backoff > maxBackoff { | |
| backoff = maxBackoff | |
| } | |
| // ±20% jitter | |
| j := time.Duration(randInt63n(int64(backoff/5) + 1)) | |
| backoff = backoff - backoff/10 + j | |
| } else { | |
| backoff = 0 | |
| } | |
| backoff := time.Duration(0) | |
| for { | |
| wait := interval | |
| if backoff > 0 { | |
| wait = backoff | |
| } | |
| timer := time.NewTimer(wait) | |
| select { | |
| case <-ctx.Done(): | |
| if !timer.Stop() { | |
| select { | |
| case <-timer.C: | |
| default: | |
| } | |
| } | |
| flushCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) | |
| _ = w.FlushAll(flushCtx) | |
| cancel() | |
| return | |
| case <-timer.C: | |
| flushCtx, cancel := context.WithTimeout(ctx, 5*time.Second) | |
| err := w.FlushOnce(flushCtx) | |
| cancel() | |
| if err != nil { | |
| if backoff == 0 { | |
| backoff = interval | |
| } else { | |
| backoff *= 2 | |
| } | |
| const maxBackoff = 30 * time.Second | |
| if backoff > maxBackoff { | |
| backoff = maxBackoff | |
| } | |
| // ±20% jitter | |
| j := time.Duration(randInt63n(int64(backoff/5) + 1)) | |
| backoff = backoff - backoff/10 + j | |
| if backoff > maxBackoff { | |
| backoff = maxBackoff | |
| } | |
| } else { | |
| backoff = 0 | |
| } |
🤖 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/flush.go` around lines 139 - 177, Re-clamp the retry
delay after applying jitter in the FlushOnce error path, using the existing
maxBackoff limit so backoff never exceeds 30 seconds. Keep the current
exponential backoff and jitter behavior, but ensure the final assigned backoff
value is bounded by maxBackoff.
1 participant
Summary
POST /v1/decisions/batchwritesdecision_batches/ aggregates / samples with idempotent(edge_node_id, batch_id)+payload_hashconflict handling.FlushWorkerusesHTTPFlushTransportwhen sync/CP is configured (stub otherwise).Test plan
go test ./internal/edgeverifier/... ./cmd/edge-verifier/...npx tsc --noEmitin control-planenode --test … decision-ingest.test.ts@coderabbitai review
Summary by CodeRabbit
New Features
Bug Fixes