Skip to content

Commit 2728cd5

Browse files
messagesgoel-blipmessagesgoel-blip
authored
Merge pull request #21 from Numeracode/feat/go-edge-hardening-pr-c
feat: Plan 8 PR C — decision batch ingest + HTTP flush
2 parents 35125ac + 6ddcdb8 commit 2728cd5

9 files changed

Lines changed: 742 additions & 24 deletions

File tree

‎cmd/edge-verifier/main.go‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -46,11 +46,12 @@ func main() {
4646
syncCancel context.CancelFunc
4747
syncCtx context.Context
4848
syncRunner *edgeverifier.SyncRunner
49+
cpClient *edgeverifier.ControlPlaneClient
4950
flushCancel context.CancelFunc
5051
bg sync.WaitGroup
5152
)
5253
if cfg.syncEnabled {
53-
snapStore, syncRunner, syncCtx, syncCancel = mustStartSync(cfg, metrics, decisionWAL)
54+
snapStore, syncRunner, cpClient, syncCtx, syncCancel = mustStartSync(cfg, metrics, decisionWAL)
5455
}
5556

5657
backendServer := startMockBackend()
@@ -64,7 +65,14 @@ func main() {
6465
if decisionWAL != nil {
6566
var flushCtx context.Context
6667
flushCtx, flushCancel = context.WithCancel(context.Background())
67-
worker := edgeverifier.NewFlushWorker(decisionWAL, &edgeverifier.StubTransport{Logger: log.Default()})
68+
var transport edgeverifier.FlushTransport = &edgeverifier.StubTransport{Logger: log.Default()}
69+
if cpClient != nil {
70+
transport = &edgeverifier.HTTPFlushTransport{Client: cpClient}
71+
log.Printf("Decision flush transport: HTTP → %s/v1/decisions/batch", cfg.controlPlaneURL)
72+
} else {
73+
log.Printf("Decision flush transport: stub (no control plane)")
74+
}
75+
worker := edgeverifier.NewFlushWorker(decisionWAL, transport)
6876
bg.Add(1)
6977
go func() {
7078
defer bg.Done()
@@ -199,7 +207,7 @@ func mustOpenWAL(cfg edgeConfig, metrics *edgeverifier.Metrics) *edgeverifier.De
199207
return wal
200208
}
201209

202-
func mustStartSync(cfg edgeConfig, metrics *edgeverifier.Metrics, wal *edgeverifier.DecisionWAL) (*edgeverifier.Store, *edgeverifier.SyncRunner, context.Context, context.CancelFunc) {
210+
func mustStartSync(cfg edgeConfig, metrics *edgeverifier.Metrics, wal *edgeverifier.DecisionWAL) (*edgeverifier.Store, *edgeverifier.SyncRunner, *edgeverifier.ControlPlaneClient, context.Context, context.CancelFunc) {
203211
if cfg.controlPlaneURL == "" || cfg.apiKey == "" {
204212
log.Fatal("sync enabled requires -control-plane-url / VERILINK_CONTROL_PLANE_URL and VERILINK_API_KEY")
205213
}
@@ -229,7 +237,7 @@ func mustStartSync(cfg edgeConfig, metrics *edgeverifier.Metrics, wal *edgeverif
229237
wal.SetNoDrop(pol.NoDropDecisions)
230238
}
231239
log.Printf("Control-plane sync enabled against %s", cfg.controlPlaneURL)
232-
return snapStore, runner, syncCtx, syncCancel
240+
return snapStore, runner, client, syncCtx, syncCancel
233241
}
234242

235243
func startMockBackend() *http.Server {

‎control-plane/src/__tests__/integration/decision-ingest.test.ts‎

Lines changed: 181 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,181 @@
1+
/**
2+
* Plan 8 PR C — decision batch ingest integration.
3+
*/
4+
process.env.DATABASE_URL ||=
5+
'postgresql://verilink:verilink@127.0.0.1:15432/verilink_test';
6+
process.env.API_KEY_HMAC_SECRET ||= 'test-hmac-secret-for-integration';
7+
8+
import { describe, it, before, after, beforeEach } from 'node:test';
9+
import assert from 'node:assert/strict';
10+
import type pg from 'pg';
11+
import { setupTestDb, teardownTestDb, resetTestData } from '../../testutil/testDb.js';
12+
import { seedTenant, seedApiKey, authHeaders } from '../../testutil/seedData.js';
13+
import { startControlPlane, type ControlPlaneHarness } from '../../testutil/appHarness.js';
14+
import {
15+
batchIDFromPayloadHash,
16+
canonicalizeDecisionsJSON,
17+
sha256Hex,
18+
shouldSample,
19+
type DecisionWire,
20+
} from '../../domains/decision/decisionSample.js';
21+
22+
function buildBatch(decisions: DecisionWire[]) {
23+
const sum = sha256Hex(canonicalizeDecisionsJSON(decisions));
24+
return {
25+
batch_id: batchIDFromPayloadHash(sum),
26+
first_wal_seq: decisions[0].wal_seq,
27+
last_wal_seq: decisions[decisions.length - 1].wal_seq,
28+
payload_hash: sum,
29+
decisions,
30+
};
31+
}
32+
33+
describe('decision batch ingest', () => {
34+
let pool: pg.Pool;
35+
let harness: ControlPlaneHarness;
36+
let apiKey: string;
37+
38+
before(async () => {
39+
pool = await setupTestDb();
40+
harness = await startControlPlane();
41+
});
42+
43+
after(async () => {
44+
await harness.stop();
45+
await teardownTestDb(pool);
46+
});
47+
48+
beforeEach(async () => {
49+
await resetTestData(pool);
50+
const tenant = await seedTenant(pool, `decisions-${Date.now()}`);
51+
apiKey = await seedApiKey(pool, tenant.id, ['*']);
52+
});
53+
54+
it('shouldSample keeps all denies and rate-limits allows', () => {
55+
const t = '2026-07-31T12:00:00.001Z';
56+
assert.equal(shouldSample('deny', t, 0), true);
57+
assert.equal(shouldSample('allow', t, 0), false);
58+
assert.equal(shouldSample('allow', t, 1), true);
59+
});
60+
61+
it('rejects payload_hash that does not match decisions', async () => {
62+
const decidedAt = new Date().toISOString();
63+
const batch = buildBatch([
64+
{
65+
wal_seq: 1,
66+
fingerprint: 'fp-a',
67+
action: 'deny',
68+
decided_at: decidedAt,
69+
},
70+
]);
71+
const resp = await fetch(`${harness.url}/v1/decisions/batch`, {
72+
method: 'POST',
73+
headers: { ...authHeaders(apiKey), 'Content-Type': 'application/json' },
74+
body: JSON.stringify({ ...batch, payload_hash: '0'.repeat(64) }),
75+
});
76+
assert.equal(resp.status, 400);
77+
});
78+
79+
it('accepts a batch and is idempotent on redelivery', async () => {
80+
const decidedAt = new Date().toISOString();
81+
const batch = buildBatch([
82+
{
83+
wal_seq: 1,
84+
fingerprint: 'fp-deny',
85+
principal_id: 'vrl:p:1',
86+
score: 10,
87+
blacklisted: false,
88+
action: 'deny',
89+
decided_at: decidedAt,
90+
},
91+
{
92+
wal_seq: 2,
93+
fingerprint: 'fp-allow',
94+
action: 'allow',
95+
score: 90,
96+
decided_at: decidedAt,
97+
},
98+
]);
99+
100+
const resp = await fetch(`${harness.url}/v1/decisions/batch`, {
101+
method: 'POST',
102+
headers: { ...authHeaders(apiKey), 'Content-Type': 'application/json' },
103+
body: JSON.stringify(batch),
104+
});
105+
assert.equal(resp.status, 200);
106+
const body = (await resp.json()) as { ok: boolean; data: { duplicate: boolean; accepted: number } };
107+
assert.equal(body.ok, true);
108+
assert.equal(body.data.duplicate, false);
109+
assert.equal(body.data.accepted, 2);
110+
111+
const again = await fetch(`${harness.url}/v1/decisions/batch`, {
112+
method: 'POST',
113+
headers: { ...authHeaders(apiKey), 'Content-Type': 'application/json' },
114+
body: JSON.stringify(batch),
115+
});
116+
assert.equal(again.status, 200);
117+
const againBody = (await again.json()) as { data: { duplicate: boolean } };
118+
assert.equal(againBody.data.duplicate, true);
119+
120+
const samples = await pool.query(`SELECT action FROM decision_samples ORDER BY wal_seq`);
121+
assert.ok(samples.rows.some((r: { action: string }) => r.action === 'deny'));
122+
123+
const aggs = await pool.query(
124+
`SELECT SUM(count)::int AS n FROM decision_aggregates WHERE dimension_kind = 'all'`
125+
);
126+
assert.equal(aggs.rows[0].n, 2);
127+
});
128+
129+
it('returns 409 when stored batch_id hash conflicts with a redelivery', async () => {
130+
const decidedAt = new Date().toISOString();
131+
const batch = buildBatch([
132+
{
133+
wal_seq: 1,
134+
fingerprint: 'fp-a',
135+
action: 'deny',
136+
decided_at: decidedAt,
137+
},
138+
]);
139+
const first = await fetch(`${harness.url}/v1/decisions/batch`, {
140+
method: 'POST',
141+
headers: { ...authHeaders(apiKey), 'Content-Type': 'application/json' },
142+
body: JSON.stringify(batch),
143+
});
144+
assert.equal(first.status, 200);
145+
146+
// Simulate a corrupted / poisoned stored hash for the same batch_id.
147+
await pool.query(`UPDATE decision_batches SET payload_hash = $1 WHERE batch_id = $2`, [
148+
'1'.repeat(64),
149+
batch.batch_id,
150+
]);
151+
152+
const resp = await fetch(`${harness.url}/v1/decisions/batch`, {
153+
method: 'POST',
154+
headers: { ...authHeaders(apiKey), 'Content-Type': 'application/json' },
155+
body: JSON.stringify(batch),
156+
});
157+
assert.equal(resp.status, 409);
158+
});
159+
160+
it('rejects non-RFC3339 decided_at', async () => {
161+
const resp = await fetch(`${harness.url}/v1/decisions/batch`, {
162+
method: 'POST',
163+
headers: { ...authHeaders(apiKey), 'Content-Type': 'application/json' },
164+
body: JSON.stringify({
165+
batch_id: '11111111-1111-1111-1111-111111111111',
166+
first_wal_seq: 1,
167+
last_wal_seq: 1,
168+
payload_hash: 'a'.repeat(64),
169+
decisions: [
170+
{
171+
wal_seq: 1,
172+
fingerprint: 'fp-a',
173+
action: 'deny',
174+
decided_at: '2026-07-31 12:00:00',
175+
},
176+
],
177+
}),
178+
});
179+
assert.equal(resp.status, 400);
180+
});
181+
});

‎control-plane/src/app.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ import principalsRouter from './routes/principals.js';
1313
import attestationsRouter from './routes/attestations.js';
1414
import syncRouter from './routes/sync.js';
1515
import adminRouter from './routes/admin.js';
16+
import decisionsRouter from './routes/decisions.js';
1617

1718
export function createApp() {
1819
const app = express();
@@ -44,6 +45,7 @@ export function createApp() {
4445
app.use('/v1/attestations', attestationsRouter);
4546
app.use('/v1/sync', syncRouter);
4647
app.use('/v1/admin', adminRouter);
48+
app.use('/v1/decisions', decisionsRouter);
4749
// Additional routes added in later plans:
4850
// app.use('/v1/policies', policiesRouter);
4951
// app.use('/v1/api-keys', apikeysRouter);

0 commit comments

Comments
 (0)