mirror of
https://github.com/garrytan/gbrain.git
synced 2026-07-27 22:15:33 +00:00
v0.40.5.0 Federated Sync v2 — parallel source sync + push triggers + per-source health (#1322)
* wip: federated sync v2 pre-merge snapshot
* v0.40.5.0 Federated Sync v2 — parallel source sync + push triggers + per-source health
Bump VERSION + package.json + CHANGELOG header + migration walkthrough filename
to v0.40.5.0 (claiming the next free slot in the v0.40.x patch series after
master's v0.40.1.0).
What ships (6 components, all behind sync.federated_v2 feature flag default-on):
1. Per-source sync lock — syncLockId(sourceId), phantom-redirect parity
2. Parallel sync --all — pMapAllSettled fan-out, --max-sources N cap
3. embed-backfill minion handler — D2 per-source lock + D6 $10/job budget + D15.1
fire-and-forget submission + D19 source-level cooldown + 24h $25 rolling cap
4. sync trigger CLI + POST /webhooks/github — HMAC-verified (60 req/min/IP),
X-GitHub-Event=push + ref filter against tracked_branch
5. sources status + federation_health doctor — batched GROUP BY pipeline
(4 queries instead of 6×N per-source roundtrips)
6. sources federate/unfederate hook — auto-submit embed-backfill on flip
Correctness fixes (unconditional):
- D21: sync.ts:959 facts backstop now passes sourceId to engine.getPage
- D15.4: redactSourceConfig + CI guard prevent webhook_secret leak
- D15.5: safeHexEqual extracted to src/core/timing-safe.ts
Schema:
- Migration v89 (sources_github_repo_index): partial expression index on
config->>'github_repo' for fast webhook source-lookup
Tests:
- 14 new test files, 112 cases. 4 IRON-RULE regressions pinned (SYNC_LOCK_ID
back-compat, phantom per-source lock, embed-backfill kill+resume,
webhook HMAC prefix-strip). All 9449 unit tests pass.
Caught at test-write time: the webhook handler had a Buffer.from('sha256=...',
'hex') truncation bug — without the prefix-strip, every signature would have
"matched" empty buffers. Pinned by a test/sources-webhook.test.ts IRON-RULE.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
* fix(check-source-config-leak): tighten regex to source-row patterns only
The v0.40.5.0 wave added scripts/check-source-config-leak.sh with a
too-broad pattern (JSON\.stringify\(.*config) that flagged any variable
named 'config' — catching the GLOBAL gbrain config.json serializers in
src/commands/init.ts (status envelopes) and src/core/config.ts (the
config-file write site). On the CI runner without rg installed, the
grep -rE fallback fired correctly and produced 4 false positives that
broke the `verify` script.
Tightened the patterns to specifically match `(source|src|row|s).config`
property access — the actual risk shape (a sources-table row being
serialized whole). The global gbrain config has a different shape and
threat model (file-mode 0o600 at the write site), so it's safe to
exempt at the regex level rather than per-file whitelist.
Also fixed a latent bug: the rg branch used `--include='*.ts'` (grep's
flag, not rg's). rg silently rejected it and CANDIDATES came back empty,
so the local-dev runs (which have rg) would never have caught a real
leak. Now branches on tool availability: `-g '*.ts'` for rg, `--include`
for grep -rE. Both branches verified against a synthetic leak fixture.
Also added init.ts + config.ts to the whitelist as a belt-and-suspenders
since they handle gbrain-global config (not source rows) and could
otherwise reflect-back via regex iteration.
CI: `bun run verify` exit 0 locally with both the original false-positive
fixture (clean repo) and a synthetic leak fixture (correctly caught,
exit 1).
---------
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Garry Tan
co-authored by
Claude Opus 4.7
Claude Opus 4.7
parent
d28be5d091
commit
df86ea5f1d
@@ -0,0 +1,86 @@
|
||||
/**
|
||||
* v0.40 Federated Sync v2 — per-source lock contract.
|
||||
*
|
||||
* Pins the iron-rule regression for the SYNC_LOCK_ID rename:
|
||||
* - SYNC_LOCK_ID === syncLockId('default') (back-compat alias)
|
||||
* - syncLockId(X) !== syncLockId(Y) for X !== Y (per-source isolation)
|
||||
* - two concurrent tryAcquireDbLock calls against DIFFERENT sources both succeed
|
||||
* - two concurrent tryAcquireDbLock calls against the SAME source: second returns null
|
||||
*
|
||||
* Without this guard, a future drift in the constant value would silently change
|
||||
* semantics (e.g. someone hardcoding 'gbrain-sync' elsewhere would no longer match
|
||||
* the same row, breaking the writer-window exclusion that performSync relies on).
|
||||
*/
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
||||
import { resetPgliteState } from './helpers/reset-pglite.ts';
|
||||
import { syncLockId, SYNC_LOCK_ID, tryAcquireDbLock } from '../src/core/db-lock.ts';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
}, 30000);
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await resetPgliteState(engine);
|
||||
});
|
||||
|
||||
describe('syncLockId (per-source lock helper)', () => {
|
||||
test('default source returns gbrain-sync:default', () => {
|
||||
expect(syncLockId('default')).toBe('gbrain-sync:default');
|
||||
});
|
||||
|
||||
test('non-default source returns gbrain-sync:<id>', () => {
|
||||
expect(syncLockId('zion-brain')).toBe('gbrain-sync:zion-brain');
|
||||
expect(syncLockId('media-corpus')).toBe('gbrain-sync:media-corpus');
|
||||
});
|
||||
|
||||
test('IRON-RULE: SYNC_LOCK_ID back-compat alias resolves to syncLockId(default)', () => {
|
||||
expect(SYNC_LOCK_ID).toBe(syncLockId('default'));
|
||||
expect(SYNC_LOCK_ID).toBe('gbrain-sync:default');
|
||||
});
|
||||
|
||||
test('different sources produce distinct lock keys', () => {
|
||||
expect(syncLockId('a')).not.toBe(syncLockId('b'));
|
||||
});
|
||||
});
|
||||
|
||||
describe('tryAcquireDbLock with per-source keys', () => {
|
||||
test('two locks against different sources both succeed', async () => {
|
||||
const lockA = await tryAcquireDbLock(engine, syncLockId('source-a'));
|
||||
const lockB = await tryAcquireDbLock(engine, syncLockId('source-b'));
|
||||
expect(lockA).not.toBeNull();
|
||||
expect(lockB).not.toBeNull();
|
||||
await lockA?.release();
|
||||
await lockB?.release();
|
||||
});
|
||||
|
||||
test('second lock against the same source returns null while first is held', async () => {
|
||||
const first = await tryAcquireDbLock(engine, syncLockId('source-c'));
|
||||
expect(first).not.toBeNull();
|
||||
const second = await tryAcquireDbLock(engine, syncLockId('source-c'));
|
||||
expect(second).toBeNull();
|
||||
await first?.release();
|
||||
// After release, third acquire succeeds.
|
||||
const third = await tryAcquireDbLock(engine, syncLockId('source-c'));
|
||||
expect(third).not.toBeNull();
|
||||
await third?.release();
|
||||
});
|
||||
|
||||
test('SYNC_LOCK_ID and syncLockId(default) acquire the SAME lock row', async () => {
|
||||
// This is the critical back-compat check: pre-v0.40 callers using SYNC_LOCK_ID
|
||||
// and new callers using syncLockId('default') MUST conflict (not bypass each other).
|
||||
const first = await tryAcquireDbLock(engine, SYNC_LOCK_ID);
|
||||
expect(first).not.toBeNull();
|
||||
const second = await tryAcquireDbLock(engine, syncLockId('default'));
|
||||
expect(second).toBeNull();
|
||||
await first?.release();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,102 @@
|
||||
/**
|
||||
* Tests for src/commands/doctor.ts:checkFederationHealth (v0.40 T12).
|
||||
*
|
||||
* Three-state contract:
|
||||
* ok — single-source brain, or every federated source healthy
|
||||
* warn — lag > 1h + federated, coverage < 95% with chunks > 100, OR 3+ failures
|
||||
* fail — lag > 24h, OR coverage < 50% with chunks > 1000
|
||||
*/
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
||||
import { checkFederationHealth } from '../src/commands/doctor.ts';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
}, 30000);
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await engine.executeRaw('DELETE FROM minion_jobs');
|
||||
await engine.executeRaw('DELETE FROM content_chunks');
|
||||
await engine.executeRaw('DELETE FROM pages');
|
||||
await engine.executeRaw(`DELETE FROM sources WHERE id != 'default'`);
|
||||
});
|
||||
|
||||
describe('checkFederationHealth', () => {
|
||||
test('single-source brain → ok with "no federation to check"', async () => {
|
||||
const check = await checkFederationHealth(engine);
|
||||
expect(check.name).toBe('federation_health');
|
||||
expect(check.status).toBe('ok');
|
||||
expect(check.message).toContain('no federation to check');
|
||||
});
|
||||
|
||||
test('multi-source healthy brain → ok with source count', async () => {
|
||||
await engine.executeRaw(
|
||||
`INSERT INTO sources (id, name, config, last_sync_at) VALUES ('extra', 'extra', '{"federated":true}', NOW())`,
|
||||
);
|
||||
const check = await checkFederationHealth(engine);
|
||||
expect(check.status).toBe('ok');
|
||||
expect(check.message).toContain('source(s) healthy');
|
||||
});
|
||||
|
||||
test('source with lag > 1h + federated → warn with remediation', async () => {
|
||||
await engine.executeRaw(
|
||||
`INSERT INTO sources (id, name, config, last_sync_at) VALUES ('stale-source', 'stale-source', '{"federated":true}', NOW() - INTERVAL '2 hours')`,
|
||||
);
|
||||
const check = await checkFederationHealth(engine);
|
||||
expect(check.status).toBe('warn');
|
||||
expect(check.message).toContain('stale-source');
|
||||
expect(check.message).toContain('gbrain sync trigger --source stale-source');
|
||||
});
|
||||
|
||||
test('source with lag > 24h → fail with remediation', async () => {
|
||||
await engine.executeRaw(
|
||||
`INSERT INTO sources (id, name, config, last_sync_at) VALUES ('dead-source', 'dead-source', '{"federated":true}', NOW() - INTERVAL '48 hours')`,
|
||||
);
|
||||
const check = await checkFederationHealth(engine);
|
||||
expect(check.status).toBe('fail');
|
||||
expect(check.message).toContain('dead-source');
|
||||
expect(check.message).toContain('gbrain sync trigger');
|
||||
});
|
||||
|
||||
test('source with low embed coverage + chunks > 100 → warn', async () => {
|
||||
await engine.executeRaw(
|
||||
`INSERT INTO sources (id, name, config, last_sync_at) VALUES ('uncovered', 'uncovered', '{"federated":true}', NOW())`,
|
||||
);
|
||||
// Seed 200 pages with chunks; 10 embedded.
|
||||
for (let i = 0; i < 200; i++) {
|
||||
await engine.putPage(`p${i}`, { type: 'note', title: `p${i}`, compiled_truth: `body ${i}` }, { sourceId: 'uncovered' });
|
||||
await engine.upsertChunks(
|
||||
`p${i}`,
|
||||
[{
|
||||
chunk_index: 0,
|
||||
chunk_text: `chunk ${i}`,
|
||||
chunk_source: 'compiled_truth',
|
||||
token_count: 1,
|
||||
embedding: i < 10 ? new Float32Array(1536) : undefined,
|
||||
}],
|
||||
{ sourceId: 'uncovered' },
|
||||
);
|
||||
}
|
||||
const check = await checkFederationHealth(engine);
|
||||
expect(check.status).toBe('warn');
|
||||
expect(check.message).toContain('uncovered');
|
||||
expect(check.message).toContain('embed coverage');
|
||||
expect(check.message).toContain('gbrain jobs submit embed-backfill');
|
||||
});
|
||||
|
||||
test('synced + zero pages → ok (vacuous truth, no coverage warn)', async () => {
|
||||
await engine.executeRaw(
|
||||
`INSERT INTO sources (id, name, config, last_sync_at) VALUES ('empty', 'empty', '{"federated":true}', NOW())`,
|
||||
);
|
||||
const check = await checkFederationHealth(engine);
|
||||
expect(check.status).toBe('ok');
|
||||
});
|
||||
});
|
||||
@@ -20,7 +20,8 @@ import { tmpdir } from 'os';
|
||||
import { hasDatabase, setupDB, teardownDB, getEngine } from './helpers.ts';
|
||||
import { withEnv } from '../helpers/with-env.ts';
|
||||
import { runExtractFacts } from '../../src/core/cycle/extract-facts.ts';
|
||||
import { tryAcquireDbLock, SYNC_LOCK_ID } from '../../src/core/db-lock.ts';
|
||||
// v0.40: per-source lock id replaces the legacy bare SYNC_LOCK_ID constant.
|
||||
// Tests below hand-craft the lock row via SQL to simulate contention.
|
||||
|
||||
const SKIP = !hasDatabase();
|
||||
const describeMaybe = SKIP ? describe.skip : describe;
|
||||
@@ -41,7 +42,8 @@ beforeEach(async () => {
|
||||
// Per-test cleanup: truncate the tables this suite mutates.
|
||||
await engine.executeRaw('TRUNCATE TABLE facts RESTART IDENTITY CASCADE');
|
||||
await engine.executeRaw('DELETE FROM pages');
|
||||
await engine.executeRaw(`DELETE FROM gbrain_cycle_locks WHERE id='gbrain-sync'`);
|
||||
// v0.40: phantom redirect now uses per-source lock id (gbrain-sync:<source>).
|
||||
await engine.executeRaw(`DELETE FROM gbrain_cycle_locks WHERE id='gbrain-sync:default'`);
|
||||
});
|
||||
|
||||
function tempBrain(): string {
|
||||
@@ -154,9 +156,10 @@ describeMaybe('phantom-redirect E2E (Postgres)', () => {
|
||||
// would still see the row as held (different "logical" holder via the
|
||||
// pid check, but the TTL is what really gates re-acquire). Override
|
||||
// by inserting a row with a different pid + future TTL.
|
||||
// v0.40 D16: phantom acquires per-source lock; default source = gbrain-sync:default.
|
||||
await engine.executeRaw(
|
||||
`INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at)
|
||||
VALUES ('gbrain-sync', 9999, 'simulated-other-host', now(), now() + interval '1 hour')
|
||||
VALUES ('gbrain-sync:default', 9999, 'simulated-other-host', now(), now() + interval '1 hour')
|
||||
ON CONFLICT (id) DO UPDATE SET holder_pid=9999, ttl_expires_at=now() + interval '1 hour'`,
|
||||
);
|
||||
|
||||
@@ -180,7 +183,8 @@ describeMaybe('phantom-redirect E2E (Postgres)', () => {
|
||||
expect(existsSync(join(brainDir, 'alice.md'))).toBe(true);
|
||||
|
||||
// Cleanup
|
||||
await engine.executeRaw(`DELETE FROM gbrain_cycle_locks WHERE id='gbrain-sync'`);
|
||||
// v0.40: phantom redirect now uses per-source lock id (gbrain-sync:<source>).
|
||||
await engine.executeRaw(`DELETE FROM gbrain_cycle_locks WHERE id='gbrain-sync:default'`);
|
||||
} finally {
|
||||
rmSync(brainDir, { recursive: true, force: true });
|
||||
}
|
||||
|
||||
@@ -0,0 +1,174 @@
|
||||
/**
|
||||
* Tests for src/core/embed-backfill-submit.ts (v0.40 D19).
|
||||
*
|
||||
* Validates the submission gate layer:
|
||||
* - Default path: submits with priority 5 + idempotency bucket
|
||||
* - Cooldown: refuses re-submission inside the window
|
||||
* - Active-job: refuses while a same-source job is active/waiting
|
||||
* - 24h spend cap: refuses when accumulated spend >= cap
|
||||
* - Config overrides honored (per-test cap + cooldown)
|
||||
* - Override knobs in opts honored (test seam)
|
||||
*/
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
||||
import { resetPgliteState } from './helpers/reset-pglite.ts';
|
||||
import {
|
||||
submitEmbedBackfill,
|
||||
COOLDOWN_CONFIG_KEY,
|
||||
SPEND_CAP_CONFIG_KEY,
|
||||
} from '../src/core/embed-backfill-submit.ts';
|
||||
import { MinionQueue } from '../src/core/minions/queue.ts';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
}, 30000); // 30s — PGLite WASM cold-start + 89 migrations exceeds 5s default
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
// Surgical reset (mirrors test/minions.test.ts) — full TRUNCATE wipes the
|
||||
// config table's `version` key that MinionQueue.ensureSchema() reads.
|
||||
await engine.executeRaw('DELETE FROM minion_jobs');
|
||||
});
|
||||
|
||||
describe('submitEmbedBackfill — happy path', () => {
|
||||
test('submits with priority 5 + idempotency key on a clean source', async () => {
|
||||
const result = await submitEmbedBackfill(engine, 'default', { reason: 'unit' });
|
||||
expect(result.status).toBe('submitted');
|
||||
expect(result.jobId).toBeDefined();
|
||||
|
||||
const queue = new MinionQueue(engine);
|
||||
const job = await queue.getJob(result.jobId!);
|
||||
expect(job).not.toBeNull();
|
||||
expect(job!.name).toBe('embed-backfill');
|
||||
expect(job!.priority).toBe(5);
|
||||
expect((job!.data as { sourceId: string }).sourceId).toBe('default');
|
||||
});
|
||||
|
||||
test('respects opts.priority override', async () => {
|
||||
const result = await submitEmbedBackfill(engine, 'default', {
|
||||
reason: 'unit',
|
||||
priority: -10,
|
||||
});
|
||||
expect(result.status).toBe('submitted');
|
||||
const queue = new MinionQueue(engine);
|
||||
const job = await queue.getJob(result.jobId!);
|
||||
expect(job!.priority).toBe(-10);
|
||||
});
|
||||
});
|
||||
|
||||
describe('submitEmbedBackfill — cooldown gate', () => {
|
||||
test('blocks re-submission while a same-source job is active', async () => {
|
||||
const queue = new MinionQueue(engine);
|
||||
// Seed an active job manually
|
||||
await queue.add('embed-backfill', { sourceId: 'default' }, {});
|
||||
await engine.executeRaw(
|
||||
`UPDATE minion_jobs SET status='active' WHERE name='embed-backfill'`,
|
||||
);
|
||||
|
||||
const result = await submitEmbedBackfill(engine, 'default', { reason: 'unit' });
|
||||
expect(result.status).toBe('cooldown');
|
||||
expect(result.cooldownRemainingSeconds).toBeUndefined();
|
||||
});
|
||||
|
||||
test('blocks re-submission inside the cooldown window after recent finish', async () => {
|
||||
const queue = new MinionQueue(engine);
|
||||
const job = await queue.add('embed-backfill', { sourceId: 'default' }, {});
|
||||
// Mark completed 1 minute ago
|
||||
await engine.executeRaw(
|
||||
`UPDATE minion_jobs SET status='completed', finished_at=NOW() - INTERVAL '1 minute' WHERE id=$1`,
|
||||
[job.id],
|
||||
);
|
||||
|
||||
const result = await submitEmbedBackfill(engine, 'default', {
|
||||
reason: 'unit',
|
||||
cooldownMinOverride: 10, // 10min cooldown; 1min elapsed → blocked
|
||||
});
|
||||
expect(result.status).toBe('cooldown');
|
||||
expect(result.cooldownRemainingSeconds).toBeGreaterThan(0);
|
||||
expect(result.cooldownRemainingSeconds).toBeLessThanOrEqual(10 * 60);
|
||||
});
|
||||
|
||||
test('allows re-submission after cooldown elapses', async () => {
|
||||
const queue = new MinionQueue(engine);
|
||||
const job = await queue.add('embed-backfill', { sourceId: 'default' }, {});
|
||||
// Mark completed 11 minutes ago — past the 10-min cooldown
|
||||
await engine.executeRaw(
|
||||
`UPDATE minion_jobs SET status='completed', finished_at=NOW() - INTERVAL '11 minutes' WHERE id=$1`,
|
||||
[job.id],
|
||||
);
|
||||
|
||||
const result = await submitEmbedBackfill(engine, 'default', {
|
||||
reason: 'unit',
|
||||
cooldownMinOverride: 10,
|
||||
});
|
||||
expect(result.status).toBe('submitted');
|
||||
});
|
||||
|
||||
test('config-overridable cooldown (via embed.backfill_cooldown_min)', async () => {
|
||||
await engine.setConfig(COOLDOWN_CONFIG_KEY, '60'); // 60min cooldown
|
||||
const queue = new MinionQueue(engine);
|
||||
const job = await queue.add('embed-backfill', { sourceId: 'default' }, {});
|
||||
await engine.executeRaw(
|
||||
`UPDATE minion_jobs SET status='completed', finished_at=NOW() - INTERVAL '30 minutes' WHERE id=$1`,
|
||||
[job.id],
|
||||
);
|
||||
|
||||
const result = await submitEmbedBackfill(engine, 'default', { reason: 'unit' });
|
||||
expect(result.status).toBe('cooldown');
|
||||
});
|
||||
});
|
||||
|
||||
describe('submitEmbedBackfill — 24h spend cap', () => {
|
||||
test('refuses when spend24hFn returns >= cap', async () => {
|
||||
const result = await submitEmbedBackfill(engine, 'default', {
|
||||
reason: 'unit',
|
||||
spendCapUsdOverride: 25,
|
||||
spend24hFn: async () => 25,
|
||||
});
|
||||
expect(result.status).toBe('spend_capped');
|
||||
expect(result.spend24hUsd).toBe(25);
|
||||
expect(result.spendCapUsd).toBe(25);
|
||||
});
|
||||
|
||||
test('admits when spend24hFn returns < cap', async () => {
|
||||
const result = await submitEmbedBackfill(engine, 'default', {
|
||||
reason: 'unit',
|
||||
spendCapUsdOverride: 25,
|
||||
spend24hFn: async () => 24.99,
|
||||
});
|
||||
expect(result.status).toBe('submitted');
|
||||
});
|
||||
|
||||
test('config-overridable spend cap (via embed.backfill_max_usd_per_source_24h)', async () => {
|
||||
await engine.setConfig(SPEND_CAP_CONFIG_KEY, '5');
|
||||
const result = await submitEmbedBackfill(engine, 'default', {
|
||||
reason: 'unit',
|
||||
spend24hFn: async () => 5,
|
||||
});
|
||||
expect(result.status).toBe('spend_capped');
|
||||
expect(result.spendCapUsd).toBe(5);
|
||||
});
|
||||
});
|
||||
|
||||
describe('submitEmbedBackfill — source isolation', () => {
|
||||
test('cooldown is per-source, not global', async () => {
|
||||
await engine.executeRaw(
|
||||
`INSERT INTO sources (id, name, config) VALUES ('other', 'other', '{"federated":true}') ON CONFLICT (id) DO NOTHING`,
|
||||
);
|
||||
const queue = new MinionQueue(engine);
|
||||
// Active job on 'default'
|
||||
await queue.add('embed-backfill', { sourceId: 'default' }, {});
|
||||
await engine.executeRaw(`UPDATE minion_jobs SET status='active' WHERE name='embed-backfill'`);
|
||||
|
||||
// Submit for 'other' — should NOT be blocked
|
||||
const result = await submitEmbedBackfill(engine, 'other', { reason: 'unit' });
|
||||
expect(result.status).toBe('submitted');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,230 @@
|
||||
/**
|
||||
* Tests for src/core/embed-stale.ts (v0.40 D15.2).
|
||||
*
|
||||
* Hermetic — uses an injected `embedFn` so no network call lands. Validates:
|
||||
* - empty stale set → done:true, embedded:0
|
||||
* - multi-batch run → embed every stale chunk, advance cursor correctly
|
||||
* - kill mid-flight (signal.aborted) → aborted:true, partial progress preserved
|
||||
* - resume from cursor → picks up where prior call left off (DB predicate)
|
||||
* - per-page embedFn throw → logged + skipped, NOT propagated; chunks stay NULL
|
||||
*
|
||||
* Why PGLite: validates the engine.listStaleChunks/getChunks/upsertChunks
|
||||
* roundtrip the helper depends on, not just the loop control flow.
|
||||
*/
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
||||
import { resetPgliteState } from './helpers/reset-pglite.ts';
|
||||
import { embedStaleForSource } from '../src/core/embed-stale.ts';
|
||||
import type { ChunkInput } from '../src/core/types.ts';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
}, 30000);
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await resetPgliteState(engine);
|
||||
});
|
||||
|
||||
/** Seed a page with N stale chunks (no embedding) into the default source. */
|
||||
async function seedPageWithStaleChunks(slug: string, chunkCount: number): Promise<void> {
|
||||
await engine.putPage(slug, {
|
||||
type: 'note',
|
||||
title: slug,
|
||||
compiled_truth: `# ${slug}\n\nseeded`,
|
||||
});
|
||||
const chunks: ChunkInput[] = Array.from({ length: chunkCount }, (_, i) => ({
|
||||
chunk_index: i,
|
||||
chunk_text: `chunk ${i} of ${slug}`,
|
||||
chunk_source: 'compiled_truth',
|
||||
token_count: 4,
|
||||
embedding: undefined, // NULL = stale
|
||||
}));
|
||||
await engine.upsertChunks(slug, chunks);
|
||||
}
|
||||
|
||||
/** Deterministic fake embedder — returns unit-length 1536-dim vectors with
|
||||
* first dim = text length, so we can assert specific chunks got embedded. */
|
||||
function fakeEmbedFn(texts: string[]): Promise<Float32Array[]> {
|
||||
return Promise.resolve(
|
||||
texts.map((t) => {
|
||||
const v = new Float32Array(1536);
|
||||
v[0] = t.length;
|
||||
v[1] = 1;
|
||||
return v;
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
describe('embedStaleForSource', () => {
|
||||
test('empty stale set returns done:true with zero embedded', async () => {
|
||||
const result = await embedStaleForSource(engine, 'default', {
|
||||
embedFn: fakeEmbedFn,
|
||||
});
|
||||
expect(result).toEqual({
|
||||
embedded: 0,
|
||||
chunksProcessed: 0,
|
||||
pagesProcessed: 0,
|
||||
lastCursor: null,
|
||||
done: true,
|
||||
aborted: false,
|
||||
});
|
||||
});
|
||||
|
||||
test('embeds every stale chunk across multiple pages in one call', async () => {
|
||||
await seedPageWithStaleChunks('a', 5);
|
||||
await seedPageWithStaleChunks('b', 3);
|
||||
|
||||
const result = await embedStaleForSource(engine, 'default', {
|
||||
embedFn: fakeEmbedFn,
|
||||
});
|
||||
expect(result.done).toBe(true);
|
||||
expect(result.aborted).toBe(false);
|
||||
expect(result.embedded).toBe(8);
|
||||
expect(result.pagesProcessed).toBe(2);
|
||||
|
||||
// Verify DB: zero stale remaining for default.
|
||||
const stale = await engine.countStaleChunks({ sourceId: 'default' });
|
||||
expect(stale).toBe(0);
|
||||
});
|
||||
|
||||
test('respects batchSize for cursor pagination', async () => {
|
||||
await seedPageWithStaleChunks('a', 3);
|
||||
await seedPageWithStaleChunks('b', 3);
|
||||
let batchCount = 0;
|
||||
const result = await embedStaleForSource(engine, 'default', {
|
||||
embedFn: fakeEmbedFn,
|
||||
batchSize: 2,
|
||||
onProgress: () => {
|
||||
batchCount++;
|
||||
},
|
||||
});
|
||||
expect(result.embedded).toBe(6);
|
||||
// 2-chunk batches across 6 stale rows = at least 3 progress callbacks.
|
||||
expect(batchCount).toBeGreaterThanOrEqual(3);
|
||||
});
|
||||
|
||||
test('IRON-RULE: aborted mid-flight → aborted:true, partial progress preserved', async () => {
|
||||
await seedPageWithStaleChunks('a', 4);
|
||||
await seedPageWithStaleChunks('b', 4);
|
||||
await seedPageWithStaleChunks('c', 4);
|
||||
const controller = new AbortController();
|
||||
// Batch size 4 = one page per batch. concurrency 1 = serialize keys.
|
||||
// Abort fires inside embedFn for page 'b', so 'a' lands, 'b' aborts mid-call,
|
||||
// and the third batch ('c') never starts.
|
||||
const result = await embedStaleForSource(engine, 'default', {
|
||||
batchSize: 4,
|
||||
concurrency: 1,
|
||||
signal: controller.signal,
|
||||
embedFn: async (texts) => {
|
||||
if (texts.some((t) => t.includes(' of b'))) {
|
||||
controller.abort();
|
||||
throw new Error('aborted'); // simulates HTTP abort throw
|
||||
}
|
||||
return fakeEmbedFn(texts);
|
||||
},
|
||||
});
|
||||
expect(result.aborted).toBe(true);
|
||||
expect(result.done).toBe(false);
|
||||
expect(result.embedded).toBe(4); // only 'a' landed
|
||||
// 'b' and 'c' (8 chunks) remain stale
|
||||
const stale = await engine.countStaleChunks({ sourceId: 'default' });
|
||||
expect(stale).toBe(8);
|
||||
});
|
||||
|
||||
test('IRON-RULE: kill + resume — second call picks up via embedding-IS-NULL predicate', async () => {
|
||||
await seedPageWithStaleChunks('a', 4);
|
||||
await seedPageWithStaleChunks('b', 4);
|
||||
|
||||
// First call aborts when 'b' is reached
|
||||
const controller = new AbortController();
|
||||
const first = await embedStaleForSource(engine, 'default', {
|
||||
batchSize: 4,
|
||||
concurrency: 1,
|
||||
signal: controller.signal,
|
||||
embedFn: async (texts) => {
|
||||
if (texts.some((t) => t.includes(' of b'))) {
|
||||
controller.abort();
|
||||
throw new Error('aborted');
|
||||
}
|
||||
return fakeEmbedFn(texts);
|
||||
},
|
||||
});
|
||||
expect(first.aborted).toBe(true);
|
||||
expect(first.embedded).toBe(4); // 'a' landed
|
||||
|
||||
// Second call with NO cursor — predicate excludes already-embedded chunks
|
||||
const second = await embedStaleForSource(engine, 'default', {
|
||||
embedFn: fakeEmbedFn,
|
||||
});
|
||||
expect(second.done).toBe(true);
|
||||
expect(first.embedded + second.embedded).toBe(8);
|
||||
|
||||
const stale = await engine.countStaleChunks({ sourceId: 'default' });
|
||||
expect(stale).toBe(0);
|
||||
});
|
||||
|
||||
test('per-page embedFn throw is logged but does NOT propagate', async () => {
|
||||
await seedPageWithStaleChunks('good', 2);
|
||||
await seedPageWithStaleChunks('bad', 2);
|
||||
|
||||
let badCount = 0;
|
||||
const result = await embedStaleForSource(engine, 'default', {
|
||||
embedFn: async (texts) => {
|
||||
if (texts.some((t) => t.includes('bad'))) {
|
||||
badCount++;
|
||||
throw new Error('intentional embed failure');
|
||||
}
|
||||
return fakeEmbedFn(texts);
|
||||
},
|
||||
});
|
||||
|
||||
// The helper itself didn't throw
|
||||
expect(result.done).toBe(true);
|
||||
expect(badCount).toBe(1);
|
||||
|
||||
// 'good' chunks got embedded; 'bad' chunks stayed NULL
|
||||
expect(result.embedded).toBe(2);
|
||||
const stale = await engine.countStaleChunks({ sourceId: 'default' });
|
||||
expect(stale).toBe(2);
|
||||
});
|
||||
|
||||
test('source-scoped: does not touch other sources', async () => {
|
||||
await engine.executeRaw(
|
||||
`INSERT INTO sources (id, name, config) VALUES ('other', 'other', '{"federated":true}'::jsonb) ON CONFLICT (id) DO NOTHING`,
|
||||
);
|
||||
await seedPageWithStaleChunks('a', 3);
|
||||
await engine.putPage('b', {
|
||||
type: 'note',
|
||||
title: 'b',
|
||||
compiled_truth: '# b\n\nseeded',
|
||||
}, { sourceId: 'other' });
|
||||
await engine.upsertChunks(
|
||||
'b',
|
||||
Array.from({ length: 3 }, (_, i) => ({
|
||||
chunk_index: i,
|
||||
chunk_text: `other ${i}`,
|
||||
chunk_source: 'compiled_truth',
|
||||
token_count: 4,
|
||||
embedding: undefined,
|
||||
})),
|
||||
{ sourceId: 'other' },
|
||||
);
|
||||
|
||||
const result = await embedStaleForSource(engine, 'default', {
|
||||
embedFn: fakeEmbedFn,
|
||||
});
|
||||
expect(result.embedded).toBe(3);
|
||||
|
||||
// 'other' source still has 3 stale chunks
|
||||
const otherStale = await engine.countStaleChunks({ sourceId: 'other' });
|
||||
expect(otherStale).toBe(3);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,52 @@
|
||||
/**
|
||||
* Tests for src/core/feature-flags.ts (v0.40 D23).
|
||||
*
|
||||
* Pin the default-on posture and the explicit-'false'-disables semantics.
|
||||
*/
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
||||
import { resetPgliteState } from './helpers/reset-pglite.ts';
|
||||
import { isFederatedV2Enabled, FEDERATED_V2_CONFIG_KEY } from '../src/core/feature-flags.ts';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
}, 30000);
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await resetPgliteState(engine);
|
||||
});
|
||||
|
||||
describe('isFederatedV2Enabled', () => {
|
||||
test('default (key unset) → enabled', async () => {
|
||||
expect(await isFederatedV2Enabled(engine)).toBe(true);
|
||||
});
|
||||
|
||||
test('explicit "false" → disabled', async () => {
|
||||
await engine.setConfig(FEDERATED_V2_CONFIG_KEY, 'false');
|
||||
expect(await isFederatedV2Enabled(engine)).toBe(false);
|
||||
});
|
||||
|
||||
test('explicit "true" → enabled', async () => {
|
||||
await engine.setConfig(FEDERATED_V2_CONFIG_KEY, 'true');
|
||||
expect(await isFederatedV2Enabled(engine)).toBe(true);
|
||||
});
|
||||
|
||||
test('anything-not-literally-false → enabled (defensive default)', async () => {
|
||||
for (const v of ['False', 'FALSE', '0', 'off', 'no', '']) {
|
||||
await engine.setConfig(FEDERATED_V2_CONFIG_KEY, v);
|
||||
expect(await isFederatedV2Enabled(engine)).toBe(true);
|
||||
}
|
||||
});
|
||||
|
||||
test('config key name is stable', () => {
|
||||
expect(FEDERATED_V2_CONFIG_KEY).toBe('sync.federated_v2');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,143 @@
|
||||
/**
|
||||
* Tests for src/core/minions/handlers/embed-backfill.ts (v0.40 D2, D6).
|
||||
*
|
||||
* Validates the handler-side contract:
|
||||
* - Happy path: embeds, returns 'success' with chunk + spend counts
|
||||
* - D2 lock: second concurrent handler call returns 'already_in_progress'
|
||||
* - D15.1 finally: lock ALWAYS releases (try/finally even on abort)
|
||||
*
|
||||
* Hermetic — uses injected embedFn via the underlying embedStaleForSource
|
||||
* test seam? No — the handler doesn't expose embedFn passthrough. Instead
|
||||
* we exercise the handler against a brain with zero stale chunks so no
|
||||
* actual embed call lands. That gives us a deterministic test of the lock
|
||||
* + budget + status branches without needing a fake gateway.
|
||||
*
|
||||
* The kill-resume contract is covered by test/embed-stale.test.ts at the
|
||||
* helper layer; the handler just routes through.
|
||||
*/
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
||||
import { makeEmbedBackfillHandler } from '../src/core/minions/handlers/embed-backfill.ts';
|
||||
import { tryAcquireDbLock } from '../src/core/db-lock.ts';
|
||||
import type { MinionJobContext } from '../src/core/minions/types.ts';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
}, 30000);
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
// Clean minion_jobs + lock rows. Preserve config (schema version + flags).
|
||||
await engine.executeRaw('DELETE FROM minion_jobs');
|
||||
await engine.executeRaw(`DELETE FROM gbrain_cycle_locks WHERE id LIKE 'gbrain-embed-backfill:%'`);
|
||||
});
|
||||
|
||||
/** Build a minimal MinionJobContext for testing. */
|
||||
function fakeJob(data: Record<string, unknown>): MinionJobContext {
|
||||
const controller = new AbortController();
|
||||
return {
|
||||
id: 1,
|
||||
name: 'embed-backfill',
|
||||
data,
|
||||
attempts_made: 0,
|
||||
signal: controller.signal,
|
||||
shutdownSignal: controller.signal,
|
||||
updateProgress: async () => {},
|
||||
updateTokens: async () => {},
|
||||
log: async () => {},
|
||||
isActive: async () => true,
|
||||
readInbox: async () => [],
|
||||
};
|
||||
}
|
||||
|
||||
describe('embed-backfill handler — happy path', () => {
|
||||
test('zero stale chunks → success with embedded=0', async () => {
|
||||
const handler = makeEmbedBackfillHandler(engine);
|
||||
const result = await handler(fakeJob({ sourceId: 'default' }));
|
||||
expect(result).toMatchObject({
|
||||
status: 'success',
|
||||
sourceId: 'default',
|
||||
embedded: 0,
|
||||
chunksProcessed: 0,
|
||||
pagesProcessed: 0,
|
||||
});
|
||||
});
|
||||
|
||||
test('throws when sourceId missing', async () => {
|
||||
const handler = makeEmbedBackfillHandler(engine);
|
||||
await expect(handler(fakeJob({}))).rejects.toThrow(/sourceId is required/);
|
||||
});
|
||||
|
||||
test('throws when sourceId is empty string', async () => {
|
||||
const handler = makeEmbedBackfillHandler(engine);
|
||||
await expect(handler(fakeJob({ sourceId: '' }))).rejects.toThrow(/sourceId is required/);
|
||||
});
|
||||
});
|
||||
|
||||
describe('embed-backfill handler — D2 lock contract', () => {
|
||||
test('IRON-RULE: second call returns already_in_progress when lock is held', async () => {
|
||||
// Hold the per-source lock externally
|
||||
const lock = await tryAcquireDbLock(engine, 'gbrain-embed-backfill:default', 60);
|
||||
expect(lock).not.toBeNull();
|
||||
|
||||
try {
|
||||
const handler = makeEmbedBackfillHandler(engine);
|
||||
const result = await handler(fakeJob({ sourceId: 'default' }));
|
||||
expect(result).toMatchObject({
|
||||
status: 'already_in_progress',
|
||||
sourceId: 'default',
|
||||
embedded: 0,
|
||||
spentUsd: 0,
|
||||
});
|
||||
} finally {
|
||||
await lock?.release();
|
||||
}
|
||||
});
|
||||
|
||||
test('different sources do not contend on each other locks', async () => {
|
||||
await engine.executeRaw(
|
||||
`INSERT INTO sources (id, name, config) VALUES ('other-src', 'other-src', '{"federated":true}') ON CONFLICT (id) DO NOTHING`,
|
||||
);
|
||||
const lockA = await tryAcquireDbLock(engine, 'gbrain-embed-backfill:default', 60);
|
||||
expect(lockA).not.toBeNull();
|
||||
try {
|
||||
// 'other-src' should still succeed
|
||||
const handler = makeEmbedBackfillHandler(engine);
|
||||
const result = await handler(fakeJob({ sourceId: 'other-src' }));
|
||||
expect(result.status).toBe('success');
|
||||
} finally {
|
||||
await lockA?.release();
|
||||
}
|
||||
});
|
||||
|
||||
test('IRON-RULE: lock is released after handler completes (try/finally)', async () => {
|
||||
const handler = makeEmbedBackfillHandler(engine);
|
||||
await handler(fakeJob({ sourceId: 'default' }));
|
||||
|
||||
// After handler returns, the lock row should NOT block a fresh acquire.
|
||||
const lock = await tryAcquireDbLock(engine, 'gbrain-embed-backfill:default', 60);
|
||||
expect(lock).not.toBeNull();
|
||||
await lock?.release();
|
||||
});
|
||||
|
||||
test('IRON-RULE: lock released on throw (sourceId-missing path)', async () => {
|
||||
const handler = makeEmbedBackfillHandler(engine);
|
||||
try {
|
||||
await handler(fakeJob({})); // throws before lock is acquired
|
||||
} catch {
|
||||
// expected
|
||||
}
|
||||
// Lock was never acquired (throw happened in parseParams pre-lock),
|
||||
// so the row should be cleanly absent. Verify a fresh acquire works.
|
||||
const lock = await tryAcquireDbLock(engine, 'gbrain-embed-backfill:default', 60);
|
||||
expect(lock).not.toBeNull();
|
||||
await lock?.release();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,83 @@
|
||||
/**
|
||||
* Tests for src/core/parallel.ts (v0.40 Federated Sync v2).
|
||||
*
|
||||
* Pins: result order matches input order, one rejection doesn't kill others,
|
||||
* concurrency cap is a hard ceiling, empty input returns empty, invalid
|
||||
* concurrency throws TypeError.
|
||||
*/
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import { pMapAllSettled } from '../src/core/parallel.ts';
|
||||
|
||||
describe('pMapAllSettled', () => {
|
||||
test('returns results in input order', async () => {
|
||||
const items = [1, 2, 3, 4, 5];
|
||||
const results = await pMapAllSettled(items, 2, async (n) => n * 10);
|
||||
expect(results).toHaveLength(5);
|
||||
expect(results.map((r) => (r.status === 'fulfilled' ? r.value : null))).toEqual([10, 20, 30, 40, 50]);
|
||||
});
|
||||
|
||||
test('one rejection does not kill other items', async () => {
|
||||
const items = ['a', 'b', 'c'];
|
||||
const results = await pMapAllSettled(items, 2, async (s) => {
|
||||
if (s === 'b') throw new Error('intentional');
|
||||
return s.toUpperCase();
|
||||
});
|
||||
expect(results[0]).toEqual({ status: 'fulfilled', value: 'A' });
|
||||
expect(results[1].status).toBe('rejected');
|
||||
expect(results[1].status === 'rejected' && (results[1].reason as Error).message).toBe('intentional');
|
||||
expect(results[2]).toEqual({ status: 'fulfilled', value: 'C' });
|
||||
});
|
||||
|
||||
test('hard ceiling: never exceeds concurrency in flight', async () => {
|
||||
let inFlight = 0;
|
||||
let peak = 0;
|
||||
const items = Array.from({ length: 20 }, (_, i) => i);
|
||||
await pMapAllSettled(items, 3, async () => {
|
||||
inFlight++;
|
||||
peak = Math.max(peak, inFlight);
|
||||
await new Promise((r) => setTimeout(r, 5));
|
||||
inFlight--;
|
||||
return 'ok';
|
||||
});
|
||||
expect(peak).toBeLessThanOrEqual(3);
|
||||
expect(peak).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
test('empty input returns empty array', async () => {
|
||||
const results = await pMapAllSettled([], 5, async () => 'x');
|
||||
expect(results).toEqual([]);
|
||||
});
|
||||
|
||||
test('concurrency > items.length runs all in parallel (no semaphore stall)', async () => {
|
||||
const start = Date.now();
|
||||
const items = [1, 2, 3];
|
||||
await pMapAllSettled(items, 100, async () => {
|
||||
await new Promise((r) => setTimeout(r, 50));
|
||||
return 'ok';
|
||||
});
|
||||
const elapsed = Date.now() - start;
|
||||
// Sequential would be ~150ms; concurrent ~50ms. Allow generous slack.
|
||||
expect(elapsed).toBeLessThan(140);
|
||||
});
|
||||
|
||||
test('throws TypeError on concurrency < 1', async () => {
|
||||
await expect(pMapAllSettled([1], 0, async (x) => x)).rejects.toThrow(TypeError);
|
||||
await expect(pMapAllSettled([1], -3, async (x) => x)).rejects.toThrow(TypeError);
|
||||
});
|
||||
|
||||
test('throws TypeError on non-integer concurrency', async () => {
|
||||
await expect(pMapAllSettled([1], 1.5, async (x) => x)).rejects.toThrow(TypeError);
|
||||
await expect(pMapAllSettled([1], NaN, async (x) => x)).rejects.toThrow(TypeError);
|
||||
});
|
||||
|
||||
test('passes index to fn', async () => {
|
||||
const seen: Array<[string, number]> = [];
|
||||
await pMapAllSettled(['a', 'b', 'c'], 2, async (item, i) => {
|
||||
seen.push([item, i]);
|
||||
return item;
|
||||
});
|
||||
// Sort by item since execution order isn't guaranteed.
|
||||
seen.sort((a, b) => a[0].localeCompare(b[0]));
|
||||
expect(seen).toEqual([['a', 0], ['b', 1], ['c', 2]]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,42 @@
|
||||
/**
|
||||
* v0.40 D16 regression test: phantom-redirect uses per-source lock.
|
||||
*
|
||||
* Pre-v0.40 phantom-redirect acquired the bare `gbrain-sync` lock, blocking
|
||||
* every concurrent sync brain-wide. After D16, phantom acquires
|
||||
* `gbrain-sync:<sourceId>` — same-source sync still serializes; cross-source
|
||||
* sync proceeds unblocked.
|
||||
*
|
||||
* Pure source-text regression guard. The full behavior is covered by
|
||||
* `test/e2e/phantom-redirect.test.ts` (DATABASE_URL-gated). This file pins
|
||||
* the lock-id construction so any future drift back to the bare constant
|
||||
* fails loudly in the fast unit loop.
|
||||
*/
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import { readFileSync } from 'node:fs';
|
||||
import { syncLockId } from '../src/core/db-lock.ts';
|
||||
|
||||
const SRC = readFileSync('src/core/cycle/phantom-redirect.ts', 'utf8');
|
||||
|
||||
describe('phantom-redirect lock contract', () => {
|
||||
test('IRON-RULE: import line uses syncLockId, not bare SYNC_LOCK_ID', () => {
|
||||
// syncLockId must be imported; SYNC_LOCK_ID must NOT be (per-source posture).
|
||||
const importLine = SRC.split('\n').find((l) =>
|
||||
l.includes("from '../db-lock.ts'") && l.includes('import'),
|
||||
);
|
||||
expect(importLine).toBeDefined();
|
||||
expect(importLine).toContain('syncLockId');
|
||||
expect(importLine).not.toMatch(/\bSYNC_LOCK_ID\b/);
|
||||
});
|
||||
|
||||
test('IRON-RULE: acquireLockWithRetry call passes syncLockId(sourceId)', () => {
|
||||
// Banned: the bare `SYNC_LOCK_ID` constant slipped back in.
|
||||
// Required: the per-source helper threaded with the active sourceId.
|
||||
expect(SRC).toMatch(/acquireLockWithRetry\s*\(\s*engine\s*,\s*syncLockId\s*\(\s*sourceId\s*\)\s*\)/);
|
||||
expect(SRC).not.toMatch(/acquireLockWithRetry\s*\(\s*engine\s*,\s*SYNC_LOCK_ID\s*\)/);
|
||||
});
|
||||
|
||||
test('helper sanity: syncLockId returns gbrain-sync:<source> shape', () => {
|
||||
expect(syncLockId('default')).toBe('gbrain-sync:default');
|
||||
expect(syncLockId('zion-brain')).toBe('gbrain-sync:zion-brain');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,54 @@
|
||||
/**
|
||||
* Tests for src/core/source-config-redact.ts (v0.40 D15.4).
|
||||
*/
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import { redactSourceConfig, hasWebhookSecret } from '../src/core/source-config-redact.ts';
|
||||
|
||||
describe('redactSourceConfig', () => {
|
||||
test('redacts webhook_secret', () => {
|
||||
const input = { federated: true, webhook_secret: 'super-secret-key', github_repo: 'a/b' };
|
||||
const out = redactSourceConfig(input);
|
||||
expect(out.webhook_secret).toBe('<redacted>');
|
||||
expect(out.federated).toBe(true);
|
||||
expect(out.github_repo).toBe('a/b');
|
||||
});
|
||||
|
||||
test('does not mutate input', () => {
|
||||
const input = { webhook_secret: 'secret' };
|
||||
redactSourceConfig(input);
|
||||
expect(input.webhook_secret).toBe('secret');
|
||||
});
|
||||
|
||||
test('returns empty object for non-object input', () => {
|
||||
expect(redactSourceConfig(null)).toEqual({});
|
||||
expect(redactSourceConfig(undefined)).toEqual({});
|
||||
expect(redactSourceConfig('a string')).toEqual({});
|
||||
expect(redactSourceConfig(['arr'])).toEqual({});
|
||||
});
|
||||
|
||||
test('preserves nested objects (not deep-redacted)', () => {
|
||||
const input = {
|
||||
webhook_secret: 'x',
|
||||
nested: { allowed: 'value' },
|
||||
};
|
||||
const out = redactSourceConfig(input);
|
||||
expect(out.webhook_secret).toBe('<redacted>');
|
||||
expect(out.nested).toEqual({ allowed: 'value' });
|
||||
});
|
||||
});
|
||||
|
||||
describe('hasWebhookSecret', () => {
|
||||
test('true when set + non-empty', () => {
|
||||
expect(hasWebhookSecret({ webhook_secret: 'x' })).toBe(true);
|
||||
});
|
||||
test('false when empty', () => {
|
||||
expect(hasWebhookSecret({ webhook_secret: '' })).toBe(false);
|
||||
});
|
||||
test('false when absent', () => {
|
||||
expect(hasWebhookSecret({})).toBe(false);
|
||||
});
|
||||
test('false on non-string values', () => {
|
||||
expect(hasWebhookSecret({ webhook_secret: 42 })).toBe(false);
|
||||
expect(hasWebhookSecret(null)).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,181 @@
|
||||
/**
|
||||
* Tests for src/core/source-health.ts (v0.40 D12 + D9 + D17).
|
||||
*
|
||||
* Validates:
|
||||
* - computeAllSourceMetrics: batched GROUP BY shape, vacuous truth for zero pages
|
||||
* - resolvePriorityLabel: high/normal/low, unknown → normal + warn-once
|
||||
* - isSourceStale: never-synced + lag-exceeded + fresh + missing local_path
|
||||
*/
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
||||
import {
|
||||
computeAllSourceMetrics,
|
||||
resolvePriorityLabel,
|
||||
resolvePriority,
|
||||
isSourceStale,
|
||||
_resetPriorityWarningsForTest,
|
||||
} from '../src/core/source-health.ts';
|
||||
import { loadAllSources } from '../src/core/sources-load.ts';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
}, 30000);
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
// Surgical reset: preserves config table (schema version).
|
||||
await engine.executeRaw('DELETE FROM minion_jobs');
|
||||
await engine.executeRaw('DELETE FROM content_chunks');
|
||||
await engine.executeRaw('DELETE FROM pages');
|
||||
await engine.executeRaw(`DELETE FROM sources WHERE id != 'default'`);
|
||||
_resetPriorityWarningsForTest();
|
||||
});
|
||||
|
||||
describe('resolvePriorityLabel', () => {
|
||||
test('recognized values', () => {
|
||||
expect(resolvePriorityLabel('s', { priority: 'high' })).toBe('high');
|
||||
expect(resolvePriorityLabel('s', { priority: 'normal' })).toBe('normal');
|
||||
expect(resolvePriorityLabel('s', { priority: 'low' })).toBe('low');
|
||||
});
|
||||
test('missing → normal silently', () => {
|
||||
expect(resolvePriorityLabel('s', {})).toBe('normal');
|
||||
expect(resolvePriorityLabel('s', null)).toBe('normal');
|
||||
});
|
||||
test('unknown values → normal with warn', () => {
|
||||
// Reroute stderr to capture
|
||||
const orig = process.stderr.write.bind(process.stderr);
|
||||
let captured = '';
|
||||
process.stderr.write = ((chunk: string | Uint8Array) => {
|
||||
captured += typeof chunk === 'string' ? chunk : Buffer.from(chunk).toString('utf8');
|
||||
return true;
|
||||
}) as never;
|
||||
try {
|
||||
expect(resolvePriorityLabel('zion-brain', { priority: 'urgent' })).toBe('normal');
|
||||
expect(captured).toContain('zion-brain');
|
||||
expect(captured).toContain('priority');
|
||||
expect(captured).toContain('normal');
|
||||
} finally {
|
||||
process.stderr.write = orig;
|
||||
}
|
||||
});
|
||||
test('warns once per source per process', () => {
|
||||
const orig = process.stderr.write.bind(process.stderr);
|
||||
let count = 0;
|
||||
process.stderr.write = ((chunk: string | Uint8Array) => {
|
||||
const s = typeof chunk === 'string' ? chunk : Buffer.from(chunk).toString('utf8');
|
||||
if (s.includes('invalid config.priority')) count++;
|
||||
return true;
|
||||
}) as never;
|
||||
try {
|
||||
resolvePriorityLabel('s1', { priority: 'urgent' });
|
||||
resolvePriorityLabel('s1', { priority: 'urgent' }); // same source
|
||||
resolvePriorityLabel('s1', { priority: 42 }); // different bad value, same source
|
||||
expect(count).toBe(1);
|
||||
} finally {
|
||||
process.stderr.write = orig;
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe('resolvePriority (numeric)', () => {
|
||||
test('maps labels to MinionQueue priority integers', () => {
|
||||
expect(resolvePriority('s', { priority: 'high' })).toBe(-10);
|
||||
expect(resolvePriority('s', { priority: 'normal' })).toBe(0);
|
||||
expect(resolvePriority('s', { priority: 'low' })).toBe(5);
|
||||
expect(resolvePriority('s', {})).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe('isSourceStale', () => {
|
||||
test('never-synced (last_sync_at null) → true', () => {
|
||||
const src = { id: 's', name: 's', local_path: '/path', last_commit: null, last_sync_at: null, config: {}, created_at: new Date() };
|
||||
expect(isSourceStale(src, 60_000)).toBe(true);
|
||||
});
|
||||
test('no local_path → false (nothing to sync)', () => {
|
||||
const src = { id: 's', name: 's', local_path: null, last_commit: null, last_sync_at: null, config: {}, created_at: new Date() };
|
||||
expect(isSourceStale(src, 60_000)).toBe(false);
|
||||
});
|
||||
test('synced within interval → false', () => {
|
||||
const src = { id: 's', name: 's', local_path: '/path', last_commit: null, last_sync_at: new Date(Date.now() - 1000), config: {}, created_at: new Date() };
|
||||
expect(isSourceStale(src, 60_000)).toBe(false);
|
||||
});
|
||||
test('synced beyond interval → true', () => {
|
||||
const src = { id: 's', name: 's', local_path: '/path', last_commit: null, last_sync_at: new Date(Date.now() - 120_000), config: {}, created_at: new Date() };
|
||||
expect(isSourceStale(src, 60_000)).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe('computeAllSourceMetrics', () => {
|
||||
test('empty input returns empty', async () => {
|
||||
const result = await computeAllSourceMetrics(engine, []);
|
||||
expect(result).toEqual([]);
|
||||
});
|
||||
|
||||
test('zero-page source → embed_coverage_pct=100 (vacuous truth)', async () => {
|
||||
const sources = await loadAllSources(engine);
|
||||
const result = await computeAllSourceMetrics(engine, sources);
|
||||
const dflt = result.find((m) => m.source_id === 'default')!;
|
||||
expect(dflt.total_pages).toBe(0);
|
||||
expect(dflt.total_chunks).toBe(0);
|
||||
expect(dflt.embed_coverage_pct).toBe(100);
|
||||
});
|
||||
|
||||
test('aggregates pages + chunks + embedding coverage per source', async () => {
|
||||
// Two pages with chunks, half embedded
|
||||
await engine.putPage('a', { type: 'note', title: 'a', compiled_truth: 'a' });
|
||||
await engine.putPage('b', { type: 'note', title: 'b', compiled_truth: 'b' });
|
||||
await engine.upsertChunks('a', [
|
||||
{ chunk_index: 0, chunk_text: 'one', chunk_source: 'compiled_truth', token_count: 1, embedding: new Float32Array(1536) },
|
||||
{ chunk_index: 1, chunk_text: 'two', chunk_source: 'compiled_truth', token_count: 1, embedding: undefined },
|
||||
]);
|
||||
await engine.upsertChunks('b', [
|
||||
{ chunk_index: 0, chunk_text: 'three', chunk_source: 'compiled_truth', token_count: 1, embedding: undefined },
|
||||
]);
|
||||
|
||||
const sources = await loadAllSources(engine);
|
||||
const result = await computeAllSourceMetrics(engine, sources);
|
||||
const dflt = result.find((m) => m.source_id === 'default')!;
|
||||
expect(dflt.total_pages).toBe(2);
|
||||
expect(dflt.total_chunks).toBe(3);
|
||||
expect(dflt.embedded_chunks).toBe(1);
|
||||
// 1/3 = 33.3%
|
||||
expect(dflt.embed_coverage_pct).toBeCloseTo(33.3, 1);
|
||||
});
|
||||
|
||||
test('lag_seconds is null when last_sync_at is null', async () => {
|
||||
const sources = await loadAllSources(engine);
|
||||
const result = await computeAllSourceMetrics(engine, sources);
|
||||
const dflt = result.find((m) => m.source_id === 'default')!;
|
||||
expect(dflt.lag_seconds).toBeNull();
|
||||
});
|
||||
|
||||
test('multi-source isolation: each source gets its own counts', async () => {
|
||||
await engine.executeRaw(`INSERT INTO sources (id, name, config) VALUES ('other', 'other', '{"federated":true}') ON CONFLICT (id) DO NOTHING`);
|
||||
await engine.putPage('a', { type: 'note', title: 'a', compiled_truth: 'a' });
|
||||
await engine.putPage('b', { type: 'note', title: 'b', compiled_truth: 'b' }, { sourceId: 'other' });
|
||||
|
||||
const sources = await loadAllSources(engine);
|
||||
const result = await computeAllSourceMetrics(engine, sources);
|
||||
expect(result.find((m) => m.source_id === 'default')!.total_pages).toBe(1);
|
||||
expect(result.find((m) => m.source_id === 'other')!.total_pages).toBe(1);
|
||||
});
|
||||
|
||||
test('webhook_configured reflects config.webhook_secret presence', async () => {
|
||||
await engine.executeRaw(
|
||||
`INSERT INTO sources (id, name, config) VALUES ('webhooky', 'webhooky', '{"federated":true,"webhook_secret":"x","github_repo":"a/b"}'::jsonb)`,
|
||||
);
|
||||
const sources = await loadAllSources(engine);
|
||||
const result = await computeAllSourceMetrics(engine, sources);
|
||||
const w = result.find((m) => m.source_id === 'webhooky')!;
|
||||
expect(w.webhook_configured).toBe(true);
|
||||
const d = result.find((m) => m.source_id === 'default')!;
|
||||
expect(d.webhook_configured).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,134 @@
|
||||
/**
|
||||
* Tests for src/core/sources-load.ts (v0.40 D7).
|
||||
*
|
||||
* Validates the shared loader: ordering, filtering, config parsing, and
|
||||
* defensive fallback when legacy brains lack the `archived` column.
|
||||
*/
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
||||
import { resetPgliteState } from './helpers/reset-pglite.ts';
|
||||
import {
|
||||
loadAllSources,
|
||||
fetchSource,
|
||||
parseSourceConfig,
|
||||
isSourceFederated,
|
||||
} from '../src/core/sources-load.ts';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
}, 30000);
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await resetPgliteState(engine);
|
||||
});
|
||||
|
||||
async function insertSource(id: string, opts: { federated?: boolean; archived?: boolean; config?: Record<string, unknown> } = {}): Promise<void> {
|
||||
const config = { ...opts.config, federated: opts.federated ?? false };
|
||||
await engine.executeRaw(
|
||||
`INSERT INTO sources (id, name, config, archived)
|
||||
VALUES ($1, $1, $2::jsonb, $3)
|
||||
ON CONFLICT (id) DO UPDATE SET config = EXCLUDED.config, archived = EXCLUDED.archived`,
|
||||
[id, JSON.stringify(config), opts.archived ?? false],
|
||||
);
|
||||
}
|
||||
|
||||
describe('loadAllSources', () => {
|
||||
test('returns default source first, then alphabetical', async () => {
|
||||
await insertSource('zebra');
|
||||
await insertSource('alpha');
|
||||
const rows = await loadAllSources(engine);
|
||||
expect(rows.map((r) => r.id)).toEqual(['default', 'alpha', 'zebra']);
|
||||
});
|
||||
|
||||
test('excludes archived rows by default', async () => {
|
||||
await insertSource('keep');
|
||||
await insertSource('archived-one', { archived: true });
|
||||
const rows = await loadAllSources(engine);
|
||||
expect(rows.map((r) => r.id).sort()).toEqual(['default', 'keep']);
|
||||
});
|
||||
|
||||
test('includeArchived: true returns archived rows', async () => {
|
||||
await insertSource('keep');
|
||||
await insertSource('archived-one', { archived: true });
|
||||
const rows = await loadAllSources(engine, { includeArchived: true });
|
||||
expect(rows.map((r) => r.id).sort()).toEqual(['archived-one', 'default', 'keep']);
|
||||
});
|
||||
|
||||
test('federatedOnly: filters to config.federated=true', async () => {
|
||||
await insertSource('iso', { federated: false });
|
||||
await insertSource('fed-a', { federated: true });
|
||||
await insertSource('fed-b', { federated: true });
|
||||
const rows = await loadAllSources(engine, { federatedOnly: true });
|
||||
// default source seeded by resetPgliteState has federated: true
|
||||
expect(rows.map((r) => r.id).sort()).toEqual(['default', 'fed-a', 'fed-b']);
|
||||
});
|
||||
|
||||
test('rows have the full SourceRow projection (no surprise drift)', async () => {
|
||||
await insertSource('shapecheck', { federated: true, config: { tracked_branch: 'main' } });
|
||||
const rows = await loadAllSources(engine);
|
||||
const target = rows.find((r) => r.id === 'shapecheck');
|
||||
expect(target).toBeDefined();
|
||||
expect(target).toMatchObject({
|
||||
id: 'shapecheck',
|
||||
name: 'shapecheck',
|
||||
local_path: null,
|
||||
last_commit: null,
|
||||
last_sync_at: null,
|
||||
});
|
||||
expect(target!.config).toBeDefined();
|
||||
expect(target!.created_at).toBeDefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('fetchSource', () => {
|
||||
test('returns row for known id', async () => {
|
||||
await insertSource('mybrain', { federated: true });
|
||||
const row = await fetchSource(engine, 'mybrain');
|
||||
expect(row?.id).toBe('mybrain');
|
||||
expect(isSourceFederated(row!.config)).toBe(true);
|
||||
});
|
||||
|
||||
test('returns null for unknown id', async () => {
|
||||
const row = await fetchSource(engine, 'does-not-exist');
|
||||
expect(row).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe('parseSourceConfig', () => {
|
||||
test('passes through plain object', () => {
|
||||
expect(parseSourceConfig({ federated: true })).toEqual({ federated: true });
|
||||
});
|
||||
|
||||
test('parses JSON string', () => {
|
||||
expect(parseSourceConfig('{"federated":true}')).toEqual({ federated: true });
|
||||
});
|
||||
|
||||
test('returns empty object on null / undefined / non-object', () => {
|
||||
expect(parseSourceConfig(null)).toEqual({});
|
||||
expect(parseSourceConfig(undefined)).toEqual({});
|
||||
expect(parseSourceConfig(42)).toEqual({});
|
||||
});
|
||||
|
||||
test('returns empty object on malformed JSON string', () => {
|
||||
expect(parseSourceConfig('{')).toEqual({});
|
||||
});
|
||||
});
|
||||
|
||||
describe('isSourceFederated', () => {
|
||||
test('strict-true requirement', () => {
|
||||
expect(isSourceFederated({ federated: true })).toBe(true);
|
||||
expect(isSourceFederated({ federated: false })).toBe(false);
|
||||
expect(isSourceFederated({ federated: 'true' })).toBe(false); // strict boolean
|
||||
expect(isSourceFederated({ federated: 1 })).toBe(false);
|
||||
expect(isSourceFederated({})).toBe(false);
|
||||
expect(isSourceFederated(null)).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,125 @@
|
||||
/**
|
||||
* Tests for /webhooks/github HMAC verification semantics (v0.40 T10).
|
||||
*
|
||||
* The handler itself is wired inside serve-http.ts's runServeHttp closure
|
||||
* (hard to invoke without bringing up the full Express app). This file pins
|
||||
* the load-bearing primitives:
|
||||
*
|
||||
* 1. GitHub HMAC sig format matches `sha256=<hex>` via createHmac.
|
||||
* 2. safeHexEqual is constant-time and rejects mismatches.
|
||||
* 3. The expected payload schema (repository.full_name + ref) parses.
|
||||
* 4. Branch ref construction: refs/heads/<tracked_branch>.
|
||||
*
|
||||
* The full HTTP path is covered by test/e2e/webhook-github.test.ts
|
||||
* (DATABASE_URL-gated; not present in this wave — filed as a follow-up
|
||||
* E2E pinned by the test plan artifact).
|
||||
*/
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import { createHmac } from 'node:crypto';
|
||||
import { safeHexEqual } from '../src/core/timing-safe.ts';
|
||||
|
||||
const GITHUB_SECRET = 'super-secret-webhook-key';
|
||||
|
||||
/** Build a sha256= HMAC sig the way GitHub does. */
|
||||
function githubSig(secret: string, payload: Buffer): string {
|
||||
return 'sha256=' + createHmac('sha256', secret).update(payload).digest('hex');
|
||||
}
|
||||
|
||||
/** Strip the GitHub prefix before constant-time hex compare. Matches the
|
||||
* webhook handler in serve-http.ts. Buffer.from('sha256=...', 'hex') silently
|
||||
* truncates at non-hex chars; comparing prefixed strings as hex returns
|
||||
* true-on-anything. The handler strips the prefix; tests must too. */
|
||||
function verifyGithubSig(secret: string, payload: Buffer, headerSig: string): boolean {
|
||||
const prefix = 'sha256=';
|
||||
if (!headerSig.startsWith(prefix)) return false;
|
||||
const expected = createHmac('sha256', secret).update(payload).digest('hex');
|
||||
return safeHexEqual(headerSig.slice(prefix.length), expected);
|
||||
}
|
||||
|
||||
/** Synthetic minimal push payload (real GitHub sends ~100KB; we only read 2 fields). */
|
||||
function pushPayload(repo: string, ref: string): Buffer {
|
||||
return Buffer.from(JSON.stringify({
|
||||
ref,
|
||||
repository: { full_name: repo, name: repo.split('/')[1], owner: { name: repo.split('/')[0] } },
|
||||
head_commit: { id: 'abc123' },
|
||||
}), 'utf8');
|
||||
}
|
||||
|
||||
describe('GitHub HMAC verification', () => {
|
||||
test('valid signature on untampered payload → verify=true', () => {
|
||||
const payload = pushPayload('Garry-s-List/zion-brain', 'refs/heads/main');
|
||||
const sig = githubSig(GITHUB_SECRET, payload);
|
||||
expect(verifyGithubSig(GITHUB_SECRET, payload, sig)).toBe(true);
|
||||
});
|
||||
|
||||
test('IRON-RULE: rejects when signature header lacks sha256= prefix', () => {
|
||||
// This was the bug: production used safeHexEqual on prefixed strings.
|
||||
// Buffer.from('sha256=...', 'hex') silently truncates at non-hex chars,
|
||||
// so both sides decoded to empty buffer and signature_mismatch never fired.
|
||||
const payload = pushPayload('owner/repo', 'refs/heads/main');
|
||||
const expected = createHmac('sha256', GITHUB_SECRET).update(payload).digest('hex');
|
||||
// Without prefix the header is invalid format → reject
|
||||
expect(verifyGithubSig(GITHUB_SECRET, payload, expected)).toBe(false);
|
||||
});
|
||||
|
||||
test('rejects when secret differs', () => {
|
||||
const payload = pushPayload('Garry-s-List/zion-brain', 'refs/heads/main');
|
||||
const goodSig = githubSig(GITHUB_SECRET, payload);
|
||||
expect(verifyGithubSig('wrong-secret', payload, goodSig)).toBe(false);
|
||||
});
|
||||
|
||||
test('rejects when payload differs (single-byte tamper)', () => {
|
||||
const payload = pushPayload('Garry-s-List/zion-brain', 'refs/heads/main');
|
||||
const sig = githubSig(GITHUB_SECRET, payload);
|
||||
const tampered = Buffer.from(payload);
|
||||
tampered[10] = tampered[10] ^ 0xff;
|
||||
expect(verifyGithubSig(GITHUB_SECRET, tampered, sig)).toBe(false);
|
||||
});
|
||||
|
||||
test('GitHub sig format is sha256=<64 hex chars>', () => {
|
||||
const payload = pushPayload('owner/repo', 'refs/heads/main');
|
||||
const sig = githubSig(GITHUB_SECRET, payload);
|
||||
expect(sig).toMatch(/^sha256=[0-9a-f]{64}$/);
|
||||
});
|
||||
});
|
||||
|
||||
describe('Push payload parsing', () => {
|
||||
test('extracts repository.full_name + ref', () => {
|
||||
const payload = pushPayload('Garry-s-List/zion-brain', 'refs/heads/main');
|
||||
const parsed = JSON.parse(payload.toString('utf8'));
|
||||
expect(parsed.repository?.full_name).toBe('Garry-s-List/zion-brain');
|
||||
expect(parsed.ref).toBe('refs/heads/main');
|
||||
});
|
||||
|
||||
test('malformed JSON throws (handler returns 400)', () => {
|
||||
const bad = Buffer.from('{not json');
|
||||
expect(() => JSON.parse(bad.toString('utf8'))).toThrow();
|
||||
});
|
||||
|
||||
test('payload missing repository.full_name is detectable', () => {
|
||||
const partial = Buffer.from(JSON.stringify({ ref: 'refs/heads/main' }));
|
||||
const parsed = JSON.parse(partial.toString('utf8'));
|
||||
expect(parsed.repository?.full_name).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('Branch ref construction (D5)', () => {
|
||||
test('default tracked_branch = main produces refs/heads/main', () => {
|
||||
const trackedBranch = 'main';
|
||||
expect(`refs/heads/${trackedBranch}`).toBe('refs/heads/main');
|
||||
});
|
||||
|
||||
test('non-main branch is exact match', () => {
|
||||
const trackedBranch: string = 'master';
|
||||
expect(`refs/heads/${trackedBranch}`).toBe('refs/heads/master');
|
||||
// ref="refs/heads/main" against tracked="master" must NOT match
|
||||
const incoming: string = 'refs/heads/main';
|
||||
expect(incoming === `refs/heads/${trackedBranch}`).toBe(false);
|
||||
});
|
||||
|
||||
test('feature-branch push to main-tracking source is rejected by exact-match', () => {
|
||||
const trackedBranch: string = 'main';
|
||||
const pushedRef: string = 'refs/heads/feature/new-stuff';
|
||||
expect(pushedRef === `refs/heads/${trackedBranch}`).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,121 @@
|
||||
/**
|
||||
* Tests for `gbrain sync trigger` CLI (v0.40 D18).
|
||||
*
|
||||
* Validates the push-trigger entry point:
|
||||
* - Help text renders
|
||||
* - --source <id> required (exits 2)
|
||||
* - --priority invalid (exits 2)
|
||||
* - Non-existent source errors before submit (exits 1)
|
||||
* - Successful submit prints job_id=N on stdout
|
||||
* - Default priority is high (-10)
|
||||
*/
|
||||
import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test';
|
||||
import { PGLiteEngine } from '../src/core/pglite-engine.ts';
|
||||
import { runSyncTrigger } from '../src/commands/sync.ts';
|
||||
import { MinionQueue } from '../src/core/minions/queue.ts';
|
||||
|
||||
let engine: PGLiteEngine;
|
||||
|
||||
beforeAll(async () => {
|
||||
engine = new PGLiteEngine();
|
||||
await engine.connect({});
|
||||
await engine.initSchema();
|
||||
}, 30000);
|
||||
|
||||
afterAll(async () => {
|
||||
await engine.disconnect();
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
await engine.executeRaw('DELETE FROM minion_jobs');
|
||||
});
|
||||
|
||||
/** Capture process.exit and stdout/stderr writes for one runSyncTrigger call. */
|
||||
async function capture(args: string[]): Promise<{
|
||||
stdout: string;
|
||||
stderr: string;
|
||||
exitCode: number | null;
|
||||
}> {
|
||||
const origExit = process.exit;
|
||||
const origLog = console.log;
|
||||
const origErr = console.error;
|
||||
let stdout = '';
|
||||
let stderr = '';
|
||||
let exitCode: number | null = null;
|
||||
const exitError = new Error('__exit__');
|
||||
process.exit = ((code?: number) => {
|
||||
exitCode = code ?? 0;
|
||||
throw exitError;
|
||||
}) as never;
|
||||
console.log = (...a: unknown[]) => { stdout += a.map(String).join(' ') + '\n'; };
|
||||
console.error = (...a: unknown[]) => { stderr += a.map(String).join(' ') + '\n'; };
|
||||
try {
|
||||
await runSyncTrigger(engine, args);
|
||||
} catch (e) {
|
||||
if (e !== exitError) throw e;
|
||||
} finally {
|
||||
process.exit = origExit;
|
||||
console.log = origLog;
|
||||
console.error = origErr;
|
||||
}
|
||||
return { stdout, stderr, exitCode };
|
||||
}
|
||||
|
||||
describe('runSyncTrigger', () => {
|
||||
test('--help prints usage and returns', async () => {
|
||||
const { stdout, exitCode } = await capture(['--help']);
|
||||
expect(exitCode).toBeNull();
|
||||
expect(stdout).toContain('gbrain sync trigger');
|
||||
expect(stdout).toContain('--source');
|
||||
expect(stdout).toContain('--priority');
|
||||
});
|
||||
|
||||
test('missing --source exits 2 with hint', async () => {
|
||||
const { stderr, exitCode } = await capture([]);
|
||||
expect(exitCode).toBe(2);
|
||||
expect(stderr).toContain('--source <id> is required');
|
||||
});
|
||||
|
||||
test('invalid --priority exits 2', async () => {
|
||||
const { stderr, exitCode } = await capture(['--source', 'default', '--priority', 'urgent']);
|
||||
expect(exitCode).toBe(2);
|
||||
expect(stderr).toContain('Invalid --priority value');
|
||||
});
|
||||
|
||||
test('non-existent source exits 1', async () => {
|
||||
const { stderr, exitCode } = await capture(['--source', 'does-not-exist']);
|
||||
expect(exitCode).toBe(1);
|
||||
expect(stderr).toContain('not found');
|
||||
});
|
||||
|
||||
test('valid trigger submits sync job + prints job_id=N to stdout', async () => {
|
||||
const { stdout, exitCode } = await capture(['--source', 'default']);
|
||||
expect(exitCode).toBeNull();
|
||||
expect(stdout).toMatch(/^job_id=\d+$/m);
|
||||
|
||||
// Verify a sync job exists with auto_embed_backfill + priority -10
|
||||
const queue = new MinionQueue(engine);
|
||||
const jobs = await queue.getJobs({ name: 'sync', limit: 5 });
|
||||
expect(jobs.length).toBe(1);
|
||||
const job = jobs[0];
|
||||
expect(job.priority).toBe(-10);
|
||||
expect((job.data as { sourceId: string }).sourceId).toBe('default');
|
||||
expect((job.data as { auto_embed_backfill: boolean }).auto_embed_backfill).toBe(true);
|
||||
});
|
||||
|
||||
test('--priority normal maps to 0', async () => {
|
||||
const { exitCode } = await capture(['--source', 'default', '--priority', 'normal']);
|
||||
expect(exitCode).toBeNull();
|
||||
const queue = new MinionQueue(engine);
|
||||
const jobs = await queue.getJobs({ name: 'sync', limit: 5 });
|
||||
expect(jobs[0].priority).toBe(0);
|
||||
});
|
||||
|
||||
test('--priority low maps to 5', async () => {
|
||||
const { exitCode } = await capture(['--source', 'default', '--priority', 'low']);
|
||||
expect(exitCode).toBeNull();
|
||||
const queue = new MinionQueue(engine);
|
||||
const jobs = await queue.getJobs({ name: 'sync', limit: 5 });
|
||||
expect(jobs[0].priority).toBe(5);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,52 @@
|
||||
/**
|
||||
* Tests for src/core/timing-safe.ts (v0.40 D15.5 extraction).
|
||||
*
|
||||
* Pure function: behavior tests plus a source-text grep guard that pins the
|
||||
* extraction so a future drift back to a serve-http-internal closure fails
|
||||
* loudly. Both the admin-cookie compare AND the webhook HMAC compare MUST
|
||||
* route through this helper.
|
||||
*/
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import { readFileSync } from 'node:fs';
|
||||
import { createHash } from 'node:crypto';
|
||||
import { safeHexEqual } from '../src/core/timing-safe.ts';
|
||||
|
||||
describe('safeHexEqual behavior', () => {
|
||||
test('returns true for identical hex strings', () => {
|
||||
const hash = createHash('sha256').update('the-quick-brown-fox').digest('hex');
|
||||
expect(safeHexEqual(hash, hash)).toBe(true);
|
||||
});
|
||||
|
||||
test('returns false for distinct hex strings of the same length', () => {
|
||||
const a = createHash('sha256').update('apple').digest('hex');
|
||||
const b = createHash('sha256').update('banana').digest('hex');
|
||||
expect(safeHexEqual(a, b)).toBe(false);
|
||||
});
|
||||
|
||||
test('returns false on length mismatch (does NOT throw)', () => {
|
||||
expect(safeHexEqual('ab', 'abcd')).toBe(false);
|
||||
expect(safeHexEqual('', 'ab')).toBe(false);
|
||||
expect(safeHexEqual('abcd', 'ab')).toBe(false);
|
||||
});
|
||||
|
||||
test('empty strings compare equal', () => {
|
||||
expect(safeHexEqual('', '')).toBe(true);
|
||||
});
|
||||
|
||||
test('Buffer.from(_, "hex") is case-insensitive — same hex bytes match either case', () => {
|
||||
// Documents Node's hex-decode behavior: 'abcd' and 'ABCD' decode to the
|
||||
// same bytes, so safeHexEqual returns true. Callers normalize beforehand
|
||||
// when they need strict case-equality.
|
||||
expect(safeHexEqual('abcd', 'ABCD')).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe('extraction contract', () => {
|
||||
test('IRON-RULE: serve-http.ts imports safeHexEqual from timing-safe.ts (not redefined)', () => {
|
||||
const src = readFileSync('src/commands/serve-http.ts', 'utf8');
|
||||
// Must import from the canonical location
|
||||
expect(src).toMatch(/import\s*\{[^}]*\bsafeHexEqual\b[^}]*\}\s*from\s*['"]\.\.\/core\/timing-safe\.ts['"]/);
|
||||
// Must NOT redefine the function inside the file
|
||||
expect(src).not.toMatch(/function\s+safeHexEqual\s*\(/);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user