bus: comment cursors survive a restart, schema 3b to 3c (row 49, #1535)
The broker stores the highest comment id seen per business and task in an append-only table, task_comment_cursors. The tasks adapter restores those ids before its startup reconcile, and the first reconcile reads each task's comment list even with no cached count, so a comment posted while the bus was down is found. Cursor rows, snapshots and comment events commit in one transaction. migrate-schema.mjs takes a stopped store from 3b to 3c offline; the store refuses a 3b database until it runs. scripts/land-work/read.mjs reads SCHEMA_VERSION from store.mjs. The live bus is 3b. From this commit on, land-work refuses it with bus-schema until Mos migrates in the schema window (agents/sage/work/window-plan/schema-chain-3c-3e.md). Candidate 87ef1321a1a08be64a20499889fc24b5549b47d6d33abfd86db95f62fb3cb6d0, Rocko packet agents/rocko/work/queue-49/r1 (ea5e70fa8), base7b166a0f1. Filbert approved in 28579 and Darkwing in 28587. Landing gate atebbbf62da: 24 suites green, 0 failed. 14 package suites passed 1598 tests (bus 178, tasks 97, business 92, cli 195, webui 112). The 10 script suites passed (land-work 84, queue 156). Follow-ups: P19 tasks 300 (duplicate cursor) and 304 (push a task's cursor only when consider runs for it). Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
Claude Opus 5.5
+39
-7
@@ -447,15 +447,47 @@ scope probes and rotation/expiry notifications; S2 never probes real services.
|
||||
|
||||
## Schema and verification
|
||||
|
||||
`schema.sql` is unchanged v3b, SHA-256
|
||||
`179ffe356d4ff19a49b5ebad39b6c6bfd7771deb8e1b7c55b5e746f039d69e65`.
|
||||
It has eight tables, **36** triggers and five views. The prototype's displayed
|
||||
35 is after deliberately dropping a guard. Source digest and a trusted full
|
||||
sqlite_master digest are checked at open, alongside schema version `3b`, WAL,
|
||||
`schema.sql` is v3c, SHA-256
|
||||
`7587ac1df813e5148fdb73dc4733ec8064e6a142e4647f0da71be009a73eebdb`.
|
||||
It has nine tables, 40 triggers and five views. Source digest and a trusted full
|
||||
sqlite_master digest are checked at open, alongside schema version `3c`, WAL,
|
||||
quick_check and foreign-key integrity. Missing metadata refuses; it never
|
||||
reinitializes an existing file. Full synchronous WAL transactions either commit
|
||||
or roll back. SQL UPDATE, DELETE and INSERT OR REPLACE refuse on every table.
|
||||
No migration or pruning command ships here.
|
||||
or roll back. SQL UPDATE, DELETE and INSERT OR REPLACE refuse on evidence tables.
|
||||
|
||||
`task_comment_cursors` holds append-only `(business, task_id, last_comment)` rows.
|
||||
The latest sequence wins, and a database guard requires the ID to increase.
|
||||
Trusted sync commits a new cursor and its comment event in one transaction;
|
||||
agent snapshots and the public task digest/projection are unchanged. An empty
|
||||
poll writes no cursor. Growth is at most one row per observed new comment, and
|
||||
usually one per batch. IDs include bots, but bot comments never contribute to
|
||||
the external-comment count. Task IDs survive project moves within a business.
|
||||
|
||||
Normal startup refuses a 3b database. The explicit offline command is:
|
||||
|
||||
```sh
|
||||
node packages/bus/src/migrate-schema.mjs /absolute/dataRoot
|
||||
```
|
||||
|
||||
Mos runs it only with the bus stopped, in an approved downtime/re-pin window,
|
||||
after a consistent backup. Development seats run it only against disposable
|
||||
fixtures. It takes the same exclusive writer lock, checks the pinned source
|
||||
schema and metadata, WAL, timestamps and integrity, then runs named migration
|
||||
steps in order. Each step atomically adds its DDL, changes only schema version
|
||||
and digest metadata, restores the metadata update guard, and verifies its target
|
||||
digest. The 3b→3c step also records an immutable `comment_cursor_epoch`. Prior
|
||||
steps stay pinned under `src/migrations/`; later versions append a step. A completed
|
||||
step remains valid if a later step refuses; rerunning resumes from that version.
|
||||
Already-current migration is a validated no-op. Missing stores refuse.
|
||||
|
||||
Before a task has a cursor, comments at or after the epoch minus 60 seconds are
|
||||
eligible (rounded down to tracker seconds). This deliberately favors a possible
|
||||
cutover duplicate over dropping comments with up to one minute of tracker clock
|
||||
lag. Older historical comments are baselined without events; a tracker clock
|
||||
more than a minute behind may still lose initial comments, and a clock ahead can
|
||||
replay already reported comments once at migration. Once an ID is stored, only
|
||||
IDs govern freshness, including after a long outage. No other event semantics,
|
||||
tracker client, or pruning command changes here.
|
||||
|
||||
Run `node --test packages/bus/tests/*.test.mjs`. Tests use private disposable
|
||||
directories, fixture token files, fake process tables, real local child brokers,
|
||||
|
||||
@@ -209,3 +209,21 @@ CREATE VIEW urgent_inbox AS
|
||||
FROM decisions d
|
||||
WHERE d.class = 'gated' AND d.blocking = 1
|
||||
AND NOT EXISTS (SELECT 1 FROM decision_events x WHERE x.decision = d.id AND x.op IN ('resolved','withdrawn','expired'));
|
||||
CREATE TABLE task_comment_cursors (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
business TEXT NOT NULL,
|
||||
task_id INTEGER NOT NULL CHECK (task_id > 0),
|
||||
last_comment INTEGER NOT NULL CHECK (last_comment > 0)
|
||||
) STRICT;
|
||||
CREATE INDEX task_comment_cursors_latest ON task_comment_cursors (business, task_id, seq);
|
||||
CREATE TRIGGER task_comment_cursors_no_update BEFORE UPDATE ON task_comment_cursors
|
||||
BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END;
|
||||
CREATE TRIGGER task_comment_cursors_no_delete BEFORE DELETE ON task_comment_cursors
|
||||
BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END;
|
||||
CREATE TRIGGER task_comment_cursors_no_replace BEFORE INSERT ON task_comment_cursors
|
||||
WHEN EXISTS (SELECT 1 FROM task_comment_cursors WHERE seq = NEW.seq)
|
||||
BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END;
|
||||
CREATE TRIGGER task_comment_cursors_forward BEFORE INSERT ON task_comment_cursors
|
||||
WHEN NEW.last_comment <= COALESCE((SELECT last_comment FROM task_comment_cursors
|
||||
WHERE business = NEW.business AND task_id = NEW.task_id ORDER BY seq DESC LIMIT 1), 0)
|
||||
BEGIN SELECT RAISE(ABORT, 'comment cursor must advance'); END;
|
||||
|
||||
+32
-12
@@ -599,23 +599,35 @@ export class Broker {
|
||||
}
|
||||
// Trusted S3 adapter only. With a cap it records a role's own completed write: the external call
|
||||
// already happened, so this records and does not re-authorize. Without one it records sync reads.
|
||||
recordTask({ cap = null, business, snapshots = [], events = [] }) {
|
||||
recordTask({ cap = null, business, snapshots = [], events = [], commentCursors = [] }) {
|
||||
const s = cap === null ? { business, role: null, run: null } : this.#session(cap);
|
||||
this.#business(business);
|
||||
if (cap !== null && (s.human || s.reader || s.business !== business)) fail('agent-required');
|
||||
if (!Array.isArray(snapshots) || !Array.isArray(events)) fail('invalid-request');
|
||||
if (!Array.isArray(snapshots) || !Array.isArray(events) || !Array.isArray(commentCursors)) fail('invalid-request');
|
||||
if (cap !== null && commentCursors.length) fail('agent-required');
|
||||
const kinds = cap === null ? POLL_KINDS : SELF_KINDS;
|
||||
this.#secretCheck({ snapshots, events });
|
||||
return this.#store.transaction(() => ({
|
||||
snapshots: snapshots.map((x) => this.#snapshot(s, x)),
|
||||
events: events.map((e) => {
|
||||
keys(e, ['kind', 'body', 'subject'], ['kind', 'body']);
|
||||
if (!kinds.has(e.kind)) fail('reserved-event');
|
||||
if (!object(e.body)) fail('invalid-request');
|
||||
return this.#event(s, e.kind, e.body, e.subject ?? null);
|
||||
}),
|
||||
}));
|
||||
this.#secretCheck({ snapshots, events, commentCursors });
|
||||
return this.#store.transaction(() => {
|
||||
for (const x of commentCursors) {
|
||||
keys(x, ['task_id', 'last_comment'], ['task_id', 'last_comment']);
|
||||
if (![x.task_id, x.last_comment].every((n) => Number.isSafeInteger(n) && n > 0)) fail('invalid-request');
|
||||
this.#store.run(
|
||||
'INSERT INTO task_comment_cursors(business,task_id,last_comment) VALUES(?,?,?)',
|
||||
business, x.task_id, x.last_comment,
|
||||
);
|
||||
}
|
||||
return {
|
||||
snapshots: snapshots.map((x) => this.#snapshot(s, x)),
|
||||
events: events.map((e) => {
|
||||
keys(e, ['kind', 'body', 'subject'], ['kind', 'body']);
|
||||
if (!kinds.has(e.kind)) fail('reserved-event');
|
||||
if (!object(e.body)) fail('invalid-request');
|
||||
return this.#event(s, e.kind, e.body, e.subject ?? null);
|
||||
}),
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
#snapshot(s, x) {
|
||||
const poll = s.role === null;
|
||||
keys(
|
||||
@@ -649,6 +661,14 @@ export class Broker {
|
||||
// priority and requirement (#taskRows).
|
||||
taskView(business, view, arg = null) {
|
||||
this.#business(business);
|
||||
if (view === 'comments' && arg === null) return {
|
||||
epoch: this.#store.get("SELECT value FROM meta WHERE key='comment_cursor_epoch'").value,
|
||||
cursors: this.#store.all(
|
||||
`SELECT task_id,last_comment FROM task_comment_cursors c WHERE business=? AND seq=(
|
||||
SELECT max(seq) FROM task_comment_cursors WHERE business=c.business AND task_id=c.task_id
|
||||
) ORDER BY task_id`, business,
|
||||
),
|
||||
};
|
||||
const parse = (r) => ({ ...r, fields: JSON.parse(r.fields) });
|
||||
if (view === 'current' && arg === null)
|
||||
return this.#store.all('SELECT * FROM task_current WHERE business=? ORDER BY task_ref', business).map(parse);
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
import { Store, SCHEMA_VERSION } from './store.mjs';
|
||||
// Offline, explicitly invoked in a human-approved downtime window. No config or secrets read.
|
||||
if (process.argv.length !== 3) {
|
||||
process.stderr.write('usage: node packages/bus/src/migrate-schema.mjs /absolute/dataRoot\n');
|
||||
process.exitCode = 2;
|
||||
} else {
|
||||
try {
|
||||
const store = new Store(process.argv[2], { migrateSchema: true });
|
||||
store.close();
|
||||
process.stdout.write(`bus schema ${SCHEMA_VERSION} ready\n`);
|
||||
} catch (e) {
|
||||
process.stderr.write(`migration refused: ${e.message}\n`);
|
||||
process.exitCode = 1;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
CREATE TABLE task_comment_cursors (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
business TEXT NOT NULL,
|
||||
task_id INTEGER NOT NULL CHECK (task_id > 0),
|
||||
last_comment INTEGER NOT NULL CHECK (last_comment > 0)
|
||||
) STRICT;
|
||||
CREATE INDEX task_comment_cursors_latest ON task_comment_cursors (business, task_id, seq);
|
||||
CREATE TRIGGER task_comment_cursors_no_update BEFORE UPDATE ON task_comment_cursors
|
||||
BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END;
|
||||
CREATE TRIGGER task_comment_cursors_no_delete BEFORE DELETE ON task_comment_cursors
|
||||
BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END;
|
||||
CREATE TRIGGER task_comment_cursors_no_replace BEFORE INSERT ON task_comment_cursors
|
||||
WHEN EXISTS (SELECT 1 FROM task_comment_cursors WHERE seq = NEW.seq)
|
||||
BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END;
|
||||
CREATE TRIGGER task_comment_cursors_forward BEFORE INSERT ON task_comment_cursors
|
||||
WHEN NEW.last_comment <= COALESCE((SELECT last_comment FROM task_comment_cursors
|
||||
WHERE business = NEW.business AND task_id = NEW.task_id ORDER BY seq DESC LIMIT 1), 0)
|
||||
BEGIN SELECT RAISE(ABORT, 'comment cursor must advance'); END;
|
||||
@@ -0,0 +1,211 @@
|
||||
PRAGMA journal_mode = WAL;
|
||||
PRAGMA foreign_keys = ON;
|
||||
CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL) STRICT;
|
||||
CREATE TABLE events (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
id TEXT NOT NULL UNIQUE,
|
||||
at TEXT NOT NULL,
|
||||
business TEXT NOT NULL,
|
||||
kind TEXT NOT NULL CHECK (kind IN (
|
||||
'session.launched',
|
||||
'session.ended',
|
||||
'action.allowed',
|
||||
'action.refused',
|
||||
'task.created',
|
||||
'task.assigned',
|
||||
'task.state',
|
||||
'task.closed',
|
||||
'task.changed.external',
|
||||
'task.conflict',
|
||||
'task.missing',
|
||||
'review.requested',
|
||||
'review.verdict',
|
||||
'human.input',
|
||||
'config.refused',
|
||||
'credential.expiring',
|
||||
'credential.expired',
|
||||
'credential.changed',
|
||||
'launch.revoked',
|
||||
'launch.restored',
|
||||
'digest.sent')),
|
||||
actor_role TEXT, actor_run TEXT,
|
||||
subject TEXT,
|
||||
corrects TEXT REFERENCES events(id),
|
||||
body TEXT NOT NULL CHECK (json_valid(body))
|
||||
) STRICT;
|
||||
CREATE TABLE role_claims (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
at TEXT NOT NULL,
|
||||
business TEXT NOT NULL, role TEXT NOT NULL,
|
||||
op TEXT NOT NULL CHECK (op IN ('claim','release','revoke')),
|
||||
holder_run TEXT NOT NULL,
|
||||
harness TEXT NOT NULL, address TEXT,
|
||||
by TEXT NOT NULL, reason TEXT,
|
||||
decision TEXT
|
||||
) STRICT;
|
||||
CREATE TABLE decisions (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
id TEXT NOT NULL UNIQUE,
|
||||
at TEXT NOT NULL,
|
||||
business TEXT NOT NULL, project TEXT,
|
||||
raised_by_role TEXT NOT NULL, raised_by_run TEXT NOT NULL,
|
||||
class TEXT NOT NULL CHECK (class IN ('routine','within-role','cross-role','gated')),
|
||||
action TEXT NOT NULL,
|
||||
route_to TEXT NOT NULL,
|
||||
question TEXT NOT NULL,
|
||||
options TEXT NOT NULL CHECK (json_valid(options) AND json_array_length(options) BETWEEN 2 AND 9),
|
||||
recommendation TEXT NOT NULL,
|
||||
task_ref TEXT CHECK (task_ref IS NULL OR (task_ref GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(task_ref, 9) GLOB '*[^0-9/]*' AND NOT substr(task_ref, 9) GLOB '*/*/*')), requirement_ref TEXT,
|
||||
blocking INTEGER NOT NULL CHECK (blocking IN (0,1)),
|
||||
supersedes TEXT REFERENCES decisions(id)
|
||||
) STRICT;
|
||||
CREATE TABLE decision_events (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
decision TEXT NOT NULL REFERENCES decisions(id),
|
||||
at TEXT NOT NULL,
|
||||
op TEXT NOT NULL CHECK (op IN ('seen','resolved','withdrawn','expired')),
|
||||
by TEXT NOT NULL,
|
||||
choice TEXT, note TEXT, via TEXT
|
||||
) STRICT;
|
||||
CREATE TABLE messages (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
id TEXT NOT NULL UNIQUE,
|
||||
at TEXT NOT NULL,
|
||||
business TEXT NOT NULL,
|
||||
from_role TEXT NOT NULL, from_run TEXT NOT NULL,
|
||||
to_role TEXT NOT NULL,
|
||||
class TEXT NOT NULL,
|
||||
in_reply_to TEXT REFERENCES messages(id),
|
||||
decision TEXT REFERENCES decisions(id),
|
||||
corrects TEXT REFERENCES messages(id),
|
||||
body TEXT NOT NULL
|
||||
) STRICT;
|
||||
CREATE TABLE deliveries (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
message TEXT NOT NULL REFERENCES messages(id),
|
||||
at TEXT NOT NULL,
|
||||
op TEXT NOT NULL CHECK (op IN ('routed','delivered','failed','read')),
|
||||
holder_run TEXT, transport TEXT, address TEXT, detail TEXT
|
||||
) STRICT;
|
||||
CREATE TABLE task_snapshots (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
at TEXT NOT NULL,
|
||||
business TEXT NOT NULL,
|
||||
task_ref TEXT NOT NULL CHECK (task_ref GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(task_ref, 9) GLOB '*[^0-9/]*' AND NOT substr(task_ref, 9) GLOB '*/*/*'),
|
||||
updated TEXT NOT NULL,
|
||||
etag TEXT,
|
||||
digest TEXT NOT NULL CHECK (length(digest) = 64 AND NOT digest GLOB '*[^0-9a-f]*'),
|
||||
fields TEXT NOT NULL CHECK (json_valid(fields)),
|
||||
source TEXT NOT NULL CHECK (source IN ('self','poll')),
|
||||
via TEXT CHECK (via IN ('board','cursor','task','reconcile')),
|
||||
read_at TEXT,
|
||||
role TEXT, run TEXT,
|
||||
CHECK ((source = 'self') = (role IS NOT NULL AND run IS NOT NULL)),
|
||||
CHECK ((source = 'poll') = (via IS NOT NULL AND read_at IS NOT NULL)),
|
||||
CHECK (json_type(fields, '$.bucket') IS 'integer'
|
||||
OR (source IS 'poll' AND via IS 'task' AND json_type(fields, '$.gone') IS 'text'))
|
||||
) STRICT;
|
||||
CREATE INDEX task_snapshots_ref ON task_snapshots (business, task_ref, seq);
|
||||
CREATE TRIGGER decisions_resolve_once BEFORE INSERT ON decision_events
|
||||
WHEN NEW.op IN ('resolved','withdrawn','expired') AND EXISTS (
|
||||
SELECT 1 FROM decision_events WHERE decision = NEW.decision AND op IN ('resolved','withdrawn','expired'))
|
||||
BEGIN SELECT RAISE(ABORT, 'decision already closed'); END;
|
||||
CREATE TRIGGER decisions_resolved_choice BEFORE INSERT ON decision_events
|
||||
WHEN NEW.op = 'resolved' AND (NEW.choice IS NULL OR NOT EXISTS (
|
||||
SELECT 1 FROM decisions d, json_each(d.options) o WHERE d.id = NEW.decision AND json_extract(o.value,'$.key') = NEW.choice))
|
||||
BEGIN SELECT RAISE(ABORT, 'resolution must name one of the options'); END;
|
||||
CREATE TRIGGER role_one_holder BEFORE INSERT ON role_claims
|
||||
WHEN NEW.op = 'claim' AND (SELECT op FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1) = 'claim'
|
||||
BEGIN SELECT RAISE(ABORT, 'role already held'); END;
|
||||
CREATE TRIGGER role_release_by_holder BEFORE INSERT ON role_claims
|
||||
WHEN NEW.op IN ('release','revoke') AND COALESCE((SELECT op FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1),'') <> 'claim'
|
||||
BEGIN SELECT RAISE(ABORT, 'role is not held'); END;
|
||||
CREATE TRIGGER role_release_same_run BEFORE INSERT ON role_claims
|
||||
WHEN NEW.op = 'release' AND (SELECT holder_run FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1) <> NEW.holder_run
|
||||
BEGIN SELECT RAISE(ABORT, 'only the holder releases; others revoke'); END;
|
||||
CREATE TRIGGER role_revoke_needs_decision BEFORE INSERT ON role_claims
|
||||
WHEN NEW.op = 'revoke' AND NEW.decision IS NULL
|
||||
BEGIN SELECT RAISE(ABORT, 'revoke needs a resolved decision'); END;
|
||||
CREATE TRIGGER meta_no_update BEFORE UPDATE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END;
|
||||
CREATE TRIGGER meta_no_delete BEFORE DELETE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END;
|
||||
CREATE TRIGGER events_no_update BEFORE UPDATE ON events BEGIN SELECT RAISE(ABORT, 'events is append-only'); END;
|
||||
CREATE TRIGGER events_no_delete BEFORE DELETE ON events BEGIN SELECT RAISE(ABORT, 'events is append-only'); END;
|
||||
CREATE TRIGGER role_claims_no_update BEFORE UPDATE ON role_claims BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END;
|
||||
CREATE TRIGGER role_claims_no_delete BEFORE DELETE ON role_claims BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END;
|
||||
CREATE TRIGGER decisions_no_update BEFORE UPDATE ON decisions BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END;
|
||||
CREATE TRIGGER decisions_no_delete BEFORE DELETE ON decisions BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END;
|
||||
CREATE TRIGGER decision_events_no_update BEFORE UPDATE ON decision_events BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END;
|
||||
CREATE TRIGGER decision_events_no_delete BEFORE DELETE ON decision_events BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END;
|
||||
CREATE TRIGGER messages_no_update BEFORE UPDATE ON messages BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END;
|
||||
CREATE TRIGGER messages_no_delete BEFORE DELETE ON messages BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END;
|
||||
CREATE TRIGGER deliveries_no_update BEFORE UPDATE ON deliveries BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END;
|
||||
CREATE TRIGGER deliveries_no_delete BEFORE DELETE ON deliveries BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END;
|
||||
CREATE TRIGGER meta_no_replace BEFORE INSERT ON meta WHEN EXISTS (SELECT 1 FROM meta WHERE key = NEW.key) BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END;
|
||||
CREATE TRIGGER events_no_replace BEFORE INSERT ON events WHEN EXISTS (SELECT 1 FROM events WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'events is append-only'); END;
|
||||
CREATE TRIGGER role_claims_no_replace BEFORE INSERT ON role_claims WHEN EXISTS (SELECT 1 FROM role_claims WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END;
|
||||
CREATE TRIGGER decisions_no_replace BEFORE INSERT ON decisions WHEN EXISTS (SELECT 1 FROM decisions WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END;
|
||||
CREATE TRIGGER decision_events_no_replace BEFORE INSERT ON decision_events WHEN EXISTS (SELECT 1 FROM decision_events WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END;
|
||||
CREATE TRIGGER messages_no_replace BEFORE INSERT ON messages WHEN EXISTS (SELECT 1 FROM messages WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END;
|
||||
CREATE TRIGGER deliveries_no_replace BEFORE INSERT ON deliveries WHEN EXISTS (SELECT 1 FROM deliveries WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END;
|
||||
CREATE TRIGGER task_snapshots_no_update BEFORE UPDATE ON task_snapshots BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END;
|
||||
CREATE TRIGGER task_snapshots_no_delete BEFORE DELETE ON task_snapshots BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END;
|
||||
CREATE TRIGGER task_snapshots_no_replace BEFORE INSERT ON task_snapshots WHEN EXISTS (SELECT 1 FROM task_snapshots WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END;
|
||||
CREATE TRIGGER decisions_blocking_needs_task BEFORE INSERT ON decisions
|
||||
WHEN NEW.blocking = 1 AND NEW.task_ref IS NULL
|
||||
BEGIN SELECT RAISE(ABORT, 'a blocking decision cites the task it blocks'); END;
|
||||
CREATE TRIGGER events_launch_by_human BEFORE INSERT ON events
|
||||
WHEN NEW.kind IN ('launch.revoked','launch.restored') AND (NEW.actor_role IS NOT NULL OR NEW.actor_run IS NOT NULL)
|
||||
BEGIN SELECT RAISE(ABORT, 'only the human revokes or restores launching'); END;
|
||||
CREATE TRIGGER events_credential_body BEFORE INSERT ON events
|
||||
WHEN NEW.kind GLOB 'credential.*' AND (
|
||||
json_extract(NEW.body, '$.service') IS NULL OR json_extract(NEW.body, '$.service') NOT IN ('gitea','vikunja')
|
||||
OR json_extract(NEW.body, '$.instance') IS NULL)
|
||||
BEGIN SELECT RAISE(ABORT, 'credential events name a service and a role instance'); END;
|
||||
CREATE TRIGGER events_task_missing_body BEFORE INSERT ON events
|
||||
WHEN NEW.kind = 'task.missing' AND (
|
||||
json_extract(NEW.body, '$.reason') IS NULL OR json_extract(NEW.body, '$.reason') NOT IN ('moved','not-found','no-access')
|
||||
OR (json_extract(NEW.body, '$.reason') = 'moved' AND json_type(NEW.body, '$.project') IS NOT 'integer'))
|
||||
BEGIN SELECT RAISE(ABORT, 'task.missing carries a reason, and the new project when moved'); END;
|
||||
CREATE TRIGGER events_task_subject BEFORE INSERT ON events
|
||||
WHEN NEW.kind GLOB 'task.*' AND (NEW.subject IS NULL OR NOT (NEW.subject GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(NEW.subject, 9) GLOB '*[^0-9/]*' AND NOT substr(NEW.subject, 9) GLOB '*/*/*'))
|
||||
BEGIN SELECT RAISE(ABORT, 'a task event names its task in subject'); END;
|
||||
CREATE TRIGGER events_task_created_body BEFORE INSERT ON events
|
||||
WHEN NEW.kind = 'task.created' AND (
|
||||
json_type(NEW.body, '$.request') IS NOT 'text'
|
||||
OR NOT EXISTS (SELECT 1 FROM events WHERE id = json_extract(NEW.body, '$.request')
|
||||
AND kind = 'human.input' AND business = NEW.business)
|
||||
OR json_type(NEW.body, '$.requirement') IS NOT 'text'
|
||||
OR NOT json_extract(NEW.body, '$.requirement') GLOB 'REQ-[A-Z]*-[1-9]*'
|
||||
OR json_extract(NEW.body, '$.requirement') GLOB 'REQ-*[^A-Z0-9-]*')
|
||||
BEGIN SELECT RAISE(ABORT, 'task.created cites the human.input that asked for it and a requirement id'); END;
|
||||
CREATE VIEW launch_state AS
|
||||
SELECT business, CASE kind WHEN 'launch.revoked' THEN 'revoked' ELSE 'allowed' END AS state, at
|
||||
FROM events e WHERE kind IN ('launch.revoked','launch.restored')
|
||||
AND seq = (SELECT max(seq) FROM events WHERE business = e.business AND kind IN ('launch.revoked','launch.restored'));
|
||||
CREATE VIEW task_external_changes AS
|
||||
SELECT p.business, p.task_ref, p.seq, p.via, p.updated, p.digest, ls.digest AS self_digest
|
||||
FROM task_snapshots p
|
||||
LEFT JOIN task_snapshots ls ON ls.seq = (SELECT max(seq) FROM task_snapshots
|
||||
WHERE business = p.business AND task_ref = p.task_ref AND source = 'self')
|
||||
WHERE p.source = 'poll'
|
||||
AND p.seq = (SELECT max(seq) FROM task_snapshots WHERE business = p.business AND task_ref = p.task_ref)
|
||||
AND (ls.seq IS NULL OR (p.updated >= ls.updated AND p.read_at > ls.at AND p.digest <> ls.digest));
|
||||
CREATE VIEW task_current AS
|
||||
SELECT s.seq, s.at, s.business, s.task_ref, s.updated, s.etag, s.digest, s.fields, s.source, s.via, s.read_at, s.role, s.run
|
||||
FROM task_snapshots s
|
||||
WHERE s.seq = (SELECT max(c.seq) FROM task_snapshots c
|
||||
WHERE c.business = s.business AND c.task_ref = s.task_ref
|
||||
AND (c.source = 'self'
|
||||
OR c.read_at > coalesce((SELECT ls.at FROM task_snapshots ls
|
||||
WHERE ls.business = c.business AND ls.task_ref = c.task_ref AND ls.source = 'self'
|
||||
ORDER BY ls.seq DESC LIMIT 1), '')));
|
||||
CREATE VIEW tasks_open AS
|
||||
SELECT s.business, s.task_ref, json_extract(s.fields, '$.bucket') AS bucket, s.seq
|
||||
FROM task_current s
|
||||
WHERE json_type(s.fields, '$.gone') IS NULL
|
||||
AND json_extract(s.fields, '$.done') = 0;
|
||||
CREATE VIEW urgent_inbox AS
|
||||
SELECT d.id, d.business, d.task_ref, d.question, d.at
|
||||
FROM decisions d
|
||||
WHERE d.class = 'gated' AND d.blocking = 1
|
||||
AND NOT EXISTS (SELECT 1 FROM decision_events x WHERE x.decision = d.id AND x.op IN ('resolved','withdrawn','expired'));
|
||||
@@ -13,7 +13,19 @@ import { join, isAbsolute } from 'node:path';
|
||||
import { createHash } from 'node:crypto';
|
||||
|
||||
const ddl = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8');
|
||||
export const SCHEMA_SHA256 = '179ffe356d4ff19a49b5ebad39b6c6bfd7771deb8e1b7c55b5e746f039d69e65';
|
||||
export const SCHEMA_SHA256 = '7587ac1df813e5148fdb73dc4733ec8064e6a142e4647f0da71be009a73eebdb';
|
||||
export const SCHEMA_VERSION = '3c';
|
||||
// Ordered, named offline steps. New schema changes append a step and keep prior DDL pinned.
|
||||
const legacyDDL = readFileSync(new URL('./migrations/3b.sql', import.meta.url), 'utf8');
|
||||
const LEGACY_SHA256 = '179ffe356d4ff19a49b5ebad39b6c6bfd7771deb8e1b7c55b5e746f039d69e65';
|
||||
const migrations = [{
|
||||
from: '3b',
|
||||
to: '3c',
|
||||
sql: readFileSync(new URL('./migrations/3b-3c.sql', import.meta.url), 'utf8'),
|
||||
initialize: (store) => store.run(
|
||||
'INSERT INTO meta(key,value) VALUES (?,?)', 'comment_cursor_epoch', new Date().toISOString(),
|
||||
),
|
||||
}];
|
||||
const timedTables = [
|
||||
'events',
|
||||
'role_claims',
|
||||
@@ -37,6 +49,19 @@ const reference = new DatabaseSync(':memory:');
|
||||
reference.exec(ddl);
|
||||
const EXPECTED_DIGEST = schemaDigest(reference);
|
||||
reference.close();
|
||||
const legacy = new DatabaseSync(':memory:');
|
||||
if (hex(legacyDDL) !== LEGACY_SHA256) throw Error('legacy-schema-source-mismatch');
|
||||
legacy.exec(legacyDDL);
|
||||
for (let i = 0; i < migrations.length; i++) {
|
||||
const step = migrations[i];
|
||||
if (i && migrations[i - 1].to !== step.from) throw Error('migration-chain-mismatch');
|
||||
step.fromDigest = schemaDigest(legacy);
|
||||
legacy.exec(step.sql);
|
||||
step.toDigest = schemaDigest(legacy);
|
||||
}
|
||||
if (migrations.at(-1).to !== SCHEMA_VERSION || schemaDigest(legacy) !== EXPECTED_DIGEST)
|
||||
throw Error('migration-target-mismatch');
|
||||
legacy.close();
|
||||
function exists(path) {
|
||||
try {
|
||||
return lstatSync(path);
|
||||
@@ -64,7 +89,7 @@ export class Store {
|
||||
#lockStat;
|
||||
#closed = false;
|
||||
#inTransaction = false;
|
||||
constructor(dataRoot) {
|
||||
constructor(dataRoot, { migrateSchema = false } = {}) {
|
||||
if (typeof dataRoot !== 'string' || !isAbsolute(dataRoot)) throw Error('unsafe-path');
|
||||
const rootStat = lstatSync(dataRoot);
|
||||
if (!rootStat.isDirectory() || rootStat.isSymbolicLink() || rootStat.uid !== process.getuid())
|
||||
@@ -94,6 +119,7 @@ export class Store {
|
||||
}
|
||||
this.path = join(this.directory, 'bus.sqlite');
|
||||
const fresh = !exists(this.path);
|
||||
if (fresh && migrateSchema) throw Error('migration-source-missing');
|
||||
if (fresh) closeSync(openSync(this.path, 'wx', 0o600));
|
||||
else safe(this.path);
|
||||
for (const suffix of ['-wal', '-shm']) if (exists(this.path + suffix)) safe(this.path + suffix);
|
||||
@@ -104,13 +130,15 @@ export class Store {
|
||||
this.#db.exec('PRAGMA journal_mode=WAL');
|
||||
this.transaction(() => {
|
||||
this.#db.exec(ddl.replace('PRAGMA journal_mode = WAL;', ''));
|
||||
this.run('INSERT INTO meta(key,value) VALUES (?,?)', 'schema_version', '3b');
|
||||
this.run('INSERT INTO meta(key,value) VALUES (?,?)', 'schema_version', SCHEMA_VERSION);
|
||||
this.run('INSERT INTO meta(key,value) VALUES (?,?)', 'schema_digest', EXPECTED_DIGEST);
|
||||
this.run('INSERT INTO meta(key,value) VALUES (?,?)', 'comment_cursor_epoch', new Date().toISOString());
|
||||
});
|
||||
}
|
||||
if (!fresh && migrateSchema) this.#migrateSchema();
|
||||
if (
|
||||
schemaDigest(this.#db) !== EXPECTED_DIGEST ||
|
||||
this.get("SELECT value FROM meta WHERE key='schema_version'")?.value !== '3b' ||
|
||||
this.get("SELECT value FROM meta WHERE key='schema_version'")?.value !== SCHEMA_VERSION ||
|
||||
this.get("SELECT value FROM meta WHERE key='schema_digest'")?.value !== EXPECTED_DIGEST ||
|
||||
this.get('PRAGMA journal_mode').journal_mode !== 'wal'
|
||||
)
|
||||
@@ -118,12 +146,41 @@ export class Store {
|
||||
if (this.get('PRAGMA quick_check').quick_check !== 'ok' || this.all('PRAGMA foreign_key_check').length)
|
||||
throw Error('integrity-failed');
|
||||
this.#verifyTimes();
|
||||
const epoch = this.get("SELECT value FROM meta WHERE key='comment_cursor_epoch'")?.value;
|
||||
if (!epoch || !Number.isFinite(Date.parse(epoch)) || new Date(epoch).toISOString() !== epoch)
|
||||
throw Error('invalid-comment-epoch');
|
||||
} catch (e) {
|
||||
if (fd !== undefined) closeSync(fd);
|
||||
this.close();
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
// Explicit offline operation only. Normal startup refuses old versions without changing them.
|
||||
#migrateSchema() {
|
||||
let version = this.get("SELECT value FROM meta WHERE key='schema_version'")?.value;
|
||||
while (version !== SCHEMA_VERSION) {
|
||||
const step = migrations.find((m) => m.from === version);
|
||||
if (
|
||||
!step || schemaDigest(this.#db) !== step.fromDigest ||
|
||||
this.get("SELECT value FROM meta WHERE key='schema_digest'")?.value !== step.fromDigest ||
|
||||
this.get('PRAGMA journal_mode').journal_mode !== 'wal'
|
||||
) throw Error('schema-mismatch');
|
||||
if (this.get('PRAGMA quick_check').quick_check !== 'ok' || this.all('PRAGMA foreign_key_check').length)
|
||||
throw Error('integrity-failed');
|
||||
this.#verifyTimes();
|
||||
this.transaction(() => {
|
||||
this.#db.exec(step.sql);
|
||||
// Only version metadata changes. The guard is restored before this step commits.
|
||||
this.#db.exec('DROP TRIGGER meta_no_update');
|
||||
this.run("UPDATE meta SET value=? WHERE key='schema_version'", step.to);
|
||||
this.run("UPDATE meta SET value=? WHERE key='schema_digest'", step.toDigest);
|
||||
step.initialize?.(this);
|
||||
this.#db.exec("CREATE TRIGGER meta_no_update BEFORE UPDATE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END");
|
||||
if (schemaDigest(this.#db) !== step.toDigest) throw Error('schema-mismatch');
|
||||
});
|
||||
version = step.to;
|
||||
}
|
||||
}
|
||||
get(sql, ...args) {
|
||||
return this.#db.prepare(sql).get(...args);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { mkdtempSync, rmSync } from 'node:fs';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import { Store } from '../src/store.mjs';
|
||||
import { Broker } from '../src/broker.mjs';
|
||||
|
||||
function fixture(t) {
|
||||
const root = mkdtempSync(join(tmpdir(), 'comment-cursors-'));
|
||||
const store = new Store(root);
|
||||
const roles = { pm: { authority: { withinRole: ['message.send'], crossRole: [] } } };
|
||||
const b = new Broker({ store, businesses: {
|
||||
demo: { id: 'demo', human: 'jason', arbiters: { technical: 'pm', delivery: 'pm' }, roles },
|
||||
other: { id: 'other', human: 'jason', arbiters: { technical: 'pm', delivery: 'pm' }, roles },
|
||||
} });
|
||||
t.after(() => { store.close(); rmSync(root, { recursive: true, force: true }); });
|
||||
return { store, b, record: (commentCursors, extra = {}) => b.recordTask({ business: 'demo', commentCursors, ...extra }) };
|
||||
}
|
||||
|
||||
test('comment cursors isolate businesses, move only forward and remain append-only', t => {
|
||||
const { store, b, record } = fixture(t);
|
||||
record([{ task_id: 7, last_comment: 11 }]);
|
||||
record([{ task_id: 7, last_comment: 15 }]);
|
||||
b.recordTask({ business: 'other', commentCursors: [{ task_id: 7, last_comment: 2 }] });
|
||||
assert.deepEqual(b.taskView('demo', 'comments').cursors.map(c => ({ ...c })), [{ task_id: 7, last_comment: 15 }]);
|
||||
assert.deepEqual(b.taskView('other', 'comments').cursors.map(c => ({ ...c })), [{ task_id: 7, last_comment: 2 }]);
|
||||
for (const last_comment of [11, 15]) assert.throws(() => record([{ task_id: 7, last_comment }]), /must advance/);
|
||||
for (const sql of [
|
||||
'UPDATE task_comment_cursors SET last_comment=20 WHERE seq=1',
|
||||
'DELETE FROM task_comment_cursors WHERE seq=1',
|
||||
"INSERT OR REPLACE INTO task_comment_cursors VALUES(1,'demo',7,20)",
|
||||
]) assert.throws(() => store.run(sql), /append-only/);
|
||||
assert.equal(store.get('SELECT count(*) n FROM task_comment_cursors').n, 3);
|
||||
});
|
||||
|
||||
test('only trusted sync writes validated cursors; events and cursor share rollback', t => {
|
||||
const { store, b, record } = fixture(t);
|
||||
const cap = b.bindLaunch({ business: 'demo', role: 'pm', run: 'pm-run', harness: 'pi' });
|
||||
for (const x of [null, {}, { task_id: 0, last_comment: 1 }, { task_id: 1, last_comment: 0 },
|
||||
{ task_id: 1, last_comment: 1.5 }, { task_id: 1, last_comment: 2 ** 53 },
|
||||
{ task_id: 1, last_comment: 1, extra: true }]) {
|
||||
assert.throws(() => record([x]), e => e.code === 'invalid-request');
|
||||
}
|
||||
assert.throws(() => record([{ task_id: 1, last_comment: 1 }], { cap }), e => e.code === 'agent-required');
|
||||
assert.throws(() => record([{ task_id: 1, last_comment: 1 }], {
|
||||
events: [{ kind: 'human.input', body: {} }],
|
||||
}), e => e.code === 'reserved-event');
|
||||
assert.equal(store.get('SELECT count(*) n FROM task_comment_cursors').n, 0);
|
||||
});
|
||||
@@ -0,0 +1,117 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { mkdtempSync, mkdirSync, readFileSync, copyFileSync, chmodSync, rmSync } from 'node:fs';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
import { DatabaseSync } from 'node:sqlite';
|
||||
import { createHash } from 'node:crypto';
|
||||
import { Store, schemaDigest } from '../src/store.mjs';
|
||||
|
||||
function legacy(t) {
|
||||
const root = mkdtempSync(join(tmpdir(), 'comment-migration-'));
|
||||
t.after(() => rmSync(root, { recursive: true, force: true }));
|
||||
const source = join(root, 'schema-current.sqlite');
|
||||
const db = new DatabaseSync(source);
|
||||
const ddl = readFileSync(new URL('./fixtures/schema-3b.sql', import.meta.url), 'utf8');
|
||||
assert.equal(createHash('sha256').update(ddl).digest('hex'),
|
||||
'179ffe356d4ff19a49b5ebad39b6c6bfd7771deb8e1b7c55b5e746f039d69e65');
|
||||
db.exec(ddl);
|
||||
db.prepare('INSERT INTO meta VALUES (?,?)').run('schema_version', '3b');
|
||||
db.prepare('INSERT INTO meta VALUES (?,?)').run('schema_digest', schemaDigest(db));
|
||||
db.prepare('INSERT INTO events(id,at,business,kind,body) VALUES(?,?,?,?,?)')
|
||||
.run('preserved', '2026-10-15T00:00:00.000Z', 'demo', 'action.allowed', '{"fixture":true}');
|
||||
db.close();
|
||||
mkdirSync(join(root, 'bus'), { mode: 0o700 });
|
||||
const path = join(root, 'bus/bus.sqlite');
|
||||
copyFileSync(source, path);
|
||||
chmodSync(path, 0o600);
|
||||
return { root, source, path };
|
||||
}
|
||||
|
||||
test('explicit 3b copy migration preserves evidence and opens as 3c; normal startup never migrates', t => {
|
||||
const { root, source } = legacy(t);
|
||||
const original = readFileSync(source);
|
||||
assert.throws(() => new Store(root), /schema-mismatch/);
|
||||
const store = new Store(root, { migrateSchema: true });
|
||||
assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3c');
|
||||
assert.equal(store.get("SELECT body FROM events WHERE id='preserved'").body, '{"fixture":true}');
|
||||
assert.equal(store.get('SELECT count(*) n FROM task_comment_cursors').n, 0);
|
||||
const epoch = store.get("SELECT value FROM meta WHERE key='comment_cursor_epoch'").value;
|
||||
assert.equal(new Date(epoch).toISOString(), epoch);
|
||||
store.close();
|
||||
assert.deepEqual(readFileSync(source), original);
|
||||
const again = new Store(root, { migrateSchema: true });
|
||||
assert.equal(again.get("SELECT value FROM meta WHERE key='comment_cursor_epoch'").value, epoch);
|
||||
again.close();
|
||||
new Store(root).close();
|
||||
});
|
||||
|
||||
test('migration refuses active writers and drifted legacy schemas, with no partial upgrade', t => {
|
||||
const { root, path } = legacy(t);
|
||||
const db = new DatabaseSync(path);
|
||||
db.exec('DROP TRIGGER events_no_delete');
|
||||
db.close();
|
||||
assert.throws(() => new Store(root, { migrateSchema: true }), /schema-mismatch/);
|
||||
const check = new DatabaseSync(path);
|
||||
assert.equal(check.prepare("SELECT value FROM meta WHERE key='schema_version'").get().value, '3b');
|
||||
assert.equal(check.prepare("SELECT count(*) n FROM sqlite_master WHERE name='task_comment_cursors'").get().n, 0);
|
||||
check.close();
|
||||
const other = legacy(t);
|
||||
const current = new Store(other.root, { migrateSchema: true });
|
||||
t.after(() => current.close());
|
||||
assert.throws(() => new Store(other.root, { migrateSchema: true }), /writer-locked/);
|
||||
});
|
||||
|
||||
test('named migration registry chains 3b through 3c to a fixture-only future version in order', async t => {
|
||||
const { root } = legacy(t);
|
||||
const { writeFileSync } = await import('node:fs');
|
||||
const { pathToFileURL } = await import('node:url');
|
||||
const moduleRoot = join(root, 'future-bus');
|
||||
mkdirSync(join(moduleRoot, 'src'), { recursive: true });
|
||||
mkdirSync(join(moduleRoot, 'src/migrations'));
|
||||
const futureSQL = 'CREATE TABLE future_fixture (value TEXT) STRICT;\n';
|
||||
const currentDDL = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8');
|
||||
const futureDDL = currentDDL + futureSQL;
|
||||
const sha = createHash('sha256').update(futureDDL).digest('hex');
|
||||
let source = readFileSync(new URL('../src/store.mjs', import.meta.url), 'utf8');
|
||||
source = source.replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${sha}';`)
|
||||
.replace("export const SCHEMA_VERSION = '3c';", "export const SCHEMA_VERSION = '3d';")
|
||||
.replace('}];\nconst timedTables', `}, { from: '3c', to: '3d', sql: ${JSON.stringify(futureSQL)} }];\nconst timedTables`);
|
||||
writeFileSync(join(moduleRoot, 'src/store.mjs'), source);
|
||||
writeFileSync(join(moduleRoot, 'schema.sql'), futureDDL);
|
||||
for (const name of ['3b.sql', '3b-3c.sql'])
|
||||
copyFileSync(new URL('../src/migrations/' + name, import.meta.url), join(moduleRoot, 'src/migrations', name));
|
||||
const { Store: FutureStore } = await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')));
|
||||
const future = new FutureStore(root, { migrateSchema: true });
|
||||
assert.equal(future.get("SELECT value FROM meta WHERE key='schema_version'").value, '3d');
|
||||
assert.equal(future.get("SELECT name FROM sqlite_master WHERE name='future_fixture'").name, 'future_fixture');
|
||||
assert.equal(future.get("SELECT count(*) n FROM task_comment_cursors").n, 0);
|
||||
assert.equal(future.get("SELECT id FROM events").id, 'preserved');
|
||||
future.close();
|
||||
new FutureStore(root).close();
|
||||
});
|
||||
|
||||
test('migration refuses mismatched source metadata without changing schema or evidence', t => {
|
||||
const { root, path } = legacy(t);
|
||||
const db = new DatabaseSync(path);
|
||||
db.exec("DROP TRIGGER meta_no_update; UPDATE meta SET value='untrusted' WHERE key='schema_digest'; CREATE TRIGGER meta_no_update BEFORE UPDATE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END");
|
||||
db.close();
|
||||
assert.throws(() => new Store(root, { migrateSchema: true }), /schema-mismatch/);
|
||||
const check = new DatabaseSync(path);
|
||||
assert.equal(check.prepare("SELECT value FROM meta WHERE key='schema_version'").get().value, '3b');
|
||||
assert.equal(check.prepare('SELECT id FROM events').get().id, 'preserved');
|
||||
check.close();
|
||||
});
|
||||
|
||||
test('offline command migrates only the supplied fixture root and reports usage errors', async t => {
|
||||
const { spawnSync } = await import('node:child_process');
|
||||
const { root } = legacy(t);
|
||||
const command = new URL('../src/migrate-schema.mjs', import.meta.url).pathname;
|
||||
const migrated = spawnSync(process.execPath, [command, root], { encoding: 'utf8' });
|
||||
assert.equal(migrated.status, 0, migrated.stderr);
|
||||
assert.match(migrated.stdout, /bus schema 3c ready/);
|
||||
new Store(root).close();
|
||||
const usage = spawnSync(process.execPath, [command], { encoding: 'utf8' });
|
||||
assert.equal(usage.status, 2);
|
||||
assert.match(usage.stderr, /usage:/);
|
||||
});
|
||||
+211
@@ -0,0 +1,211 @@
|
||||
PRAGMA journal_mode = WAL;
|
||||
PRAGMA foreign_keys = ON;
|
||||
CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL) STRICT;
|
||||
CREATE TABLE events (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
id TEXT NOT NULL UNIQUE,
|
||||
at TEXT NOT NULL,
|
||||
business TEXT NOT NULL,
|
||||
kind TEXT NOT NULL CHECK (kind IN (
|
||||
'session.launched',
|
||||
'session.ended',
|
||||
'action.allowed',
|
||||
'action.refused',
|
||||
'task.created',
|
||||
'task.assigned',
|
||||
'task.state',
|
||||
'task.closed',
|
||||
'task.changed.external',
|
||||
'task.conflict',
|
||||
'task.missing',
|
||||
'review.requested',
|
||||
'review.verdict',
|
||||
'human.input',
|
||||
'config.refused',
|
||||
'credential.expiring',
|
||||
'credential.expired',
|
||||
'credential.changed',
|
||||
'launch.revoked',
|
||||
'launch.restored',
|
||||
'digest.sent')),
|
||||
actor_role TEXT, actor_run TEXT,
|
||||
subject TEXT,
|
||||
corrects TEXT REFERENCES events(id),
|
||||
body TEXT NOT NULL CHECK (json_valid(body))
|
||||
) STRICT;
|
||||
CREATE TABLE role_claims (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
at TEXT NOT NULL,
|
||||
business TEXT NOT NULL, role TEXT NOT NULL,
|
||||
op TEXT NOT NULL CHECK (op IN ('claim','release','revoke')),
|
||||
holder_run TEXT NOT NULL,
|
||||
harness TEXT NOT NULL, address TEXT,
|
||||
by TEXT NOT NULL, reason TEXT,
|
||||
decision TEXT
|
||||
) STRICT;
|
||||
CREATE TABLE decisions (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
id TEXT NOT NULL UNIQUE,
|
||||
at TEXT NOT NULL,
|
||||
business TEXT NOT NULL, project TEXT,
|
||||
raised_by_role TEXT NOT NULL, raised_by_run TEXT NOT NULL,
|
||||
class TEXT NOT NULL CHECK (class IN ('routine','within-role','cross-role','gated')),
|
||||
action TEXT NOT NULL,
|
||||
route_to TEXT NOT NULL,
|
||||
question TEXT NOT NULL,
|
||||
options TEXT NOT NULL CHECK (json_valid(options) AND json_array_length(options) BETWEEN 2 AND 9),
|
||||
recommendation TEXT NOT NULL,
|
||||
task_ref TEXT CHECK (task_ref IS NULL OR (task_ref GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(task_ref, 9) GLOB '*[^0-9/]*' AND NOT substr(task_ref, 9) GLOB '*/*/*')), requirement_ref TEXT,
|
||||
blocking INTEGER NOT NULL CHECK (blocking IN (0,1)),
|
||||
supersedes TEXT REFERENCES decisions(id)
|
||||
) STRICT;
|
||||
CREATE TABLE decision_events (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
decision TEXT NOT NULL REFERENCES decisions(id),
|
||||
at TEXT NOT NULL,
|
||||
op TEXT NOT NULL CHECK (op IN ('seen','resolved','withdrawn','expired')),
|
||||
by TEXT NOT NULL,
|
||||
choice TEXT, note TEXT, via TEXT
|
||||
) STRICT;
|
||||
CREATE TABLE messages (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
id TEXT NOT NULL UNIQUE,
|
||||
at TEXT NOT NULL,
|
||||
business TEXT NOT NULL,
|
||||
from_role TEXT NOT NULL, from_run TEXT NOT NULL,
|
||||
to_role TEXT NOT NULL,
|
||||
class TEXT NOT NULL,
|
||||
in_reply_to TEXT REFERENCES messages(id),
|
||||
decision TEXT REFERENCES decisions(id),
|
||||
corrects TEXT REFERENCES messages(id),
|
||||
body TEXT NOT NULL
|
||||
) STRICT;
|
||||
CREATE TABLE deliveries (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
message TEXT NOT NULL REFERENCES messages(id),
|
||||
at TEXT NOT NULL,
|
||||
op TEXT NOT NULL CHECK (op IN ('routed','delivered','failed','read')),
|
||||
holder_run TEXT, transport TEXT, address TEXT, detail TEXT
|
||||
) STRICT;
|
||||
CREATE TABLE task_snapshots (
|
||||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
at TEXT NOT NULL,
|
||||
business TEXT NOT NULL,
|
||||
task_ref TEXT NOT NULL CHECK (task_ref GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(task_ref, 9) GLOB '*[^0-9/]*' AND NOT substr(task_ref, 9) GLOB '*/*/*'),
|
||||
updated TEXT NOT NULL,
|
||||
etag TEXT,
|
||||
digest TEXT NOT NULL CHECK (length(digest) = 64 AND NOT digest GLOB '*[^0-9a-f]*'),
|
||||
fields TEXT NOT NULL CHECK (json_valid(fields)),
|
||||
source TEXT NOT NULL CHECK (source IN ('self','poll')),
|
||||
via TEXT CHECK (via IN ('board','cursor','task','reconcile')),
|
||||
read_at TEXT,
|
||||
role TEXT, run TEXT,
|
||||
CHECK ((source = 'self') = (role IS NOT NULL AND run IS NOT NULL)),
|
||||
CHECK ((source = 'poll') = (via IS NOT NULL AND read_at IS NOT NULL)),
|
||||
CHECK (json_type(fields, '$.bucket') IS 'integer'
|
||||
OR (source IS 'poll' AND via IS 'task' AND json_type(fields, '$.gone') IS 'text'))
|
||||
) STRICT;
|
||||
CREATE INDEX task_snapshots_ref ON task_snapshots (business, task_ref, seq);
|
||||
CREATE TRIGGER decisions_resolve_once BEFORE INSERT ON decision_events
|
||||
WHEN NEW.op IN ('resolved','withdrawn','expired') AND EXISTS (
|
||||
SELECT 1 FROM decision_events WHERE decision = NEW.decision AND op IN ('resolved','withdrawn','expired'))
|
||||
BEGIN SELECT RAISE(ABORT, 'decision already closed'); END;
|
||||
CREATE TRIGGER decisions_resolved_choice BEFORE INSERT ON decision_events
|
||||
WHEN NEW.op = 'resolved' AND (NEW.choice IS NULL OR NOT EXISTS (
|
||||
SELECT 1 FROM decisions d, json_each(d.options) o WHERE d.id = NEW.decision AND json_extract(o.value,'$.key') = NEW.choice))
|
||||
BEGIN SELECT RAISE(ABORT, 'resolution must name one of the options'); END;
|
||||
CREATE TRIGGER role_one_holder BEFORE INSERT ON role_claims
|
||||
WHEN NEW.op = 'claim' AND (SELECT op FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1) = 'claim'
|
||||
BEGIN SELECT RAISE(ABORT, 'role already held'); END;
|
||||
CREATE TRIGGER role_release_by_holder BEFORE INSERT ON role_claims
|
||||
WHEN NEW.op IN ('release','revoke') AND COALESCE((SELECT op FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1),'') <> 'claim'
|
||||
BEGIN SELECT RAISE(ABORT, 'role is not held'); END;
|
||||
CREATE TRIGGER role_release_same_run BEFORE INSERT ON role_claims
|
||||
WHEN NEW.op = 'release' AND (SELECT holder_run FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1) <> NEW.holder_run
|
||||
BEGIN SELECT RAISE(ABORT, 'only the holder releases; others revoke'); END;
|
||||
CREATE TRIGGER role_revoke_needs_decision BEFORE INSERT ON role_claims
|
||||
WHEN NEW.op = 'revoke' AND NEW.decision IS NULL
|
||||
BEGIN SELECT RAISE(ABORT, 'revoke needs a resolved decision'); END;
|
||||
CREATE TRIGGER meta_no_update BEFORE UPDATE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END;
|
||||
CREATE TRIGGER meta_no_delete BEFORE DELETE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END;
|
||||
CREATE TRIGGER events_no_update BEFORE UPDATE ON events BEGIN SELECT RAISE(ABORT, 'events is append-only'); END;
|
||||
CREATE TRIGGER events_no_delete BEFORE DELETE ON events BEGIN SELECT RAISE(ABORT, 'events is append-only'); END;
|
||||
CREATE TRIGGER role_claims_no_update BEFORE UPDATE ON role_claims BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END;
|
||||
CREATE TRIGGER role_claims_no_delete BEFORE DELETE ON role_claims BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END;
|
||||
CREATE TRIGGER decisions_no_update BEFORE UPDATE ON decisions BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END;
|
||||
CREATE TRIGGER decisions_no_delete BEFORE DELETE ON decisions BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END;
|
||||
CREATE TRIGGER decision_events_no_update BEFORE UPDATE ON decision_events BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END;
|
||||
CREATE TRIGGER decision_events_no_delete BEFORE DELETE ON decision_events BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END;
|
||||
CREATE TRIGGER messages_no_update BEFORE UPDATE ON messages BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END;
|
||||
CREATE TRIGGER messages_no_delete BEFORE DELETE ON messages BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END;
|
||||
CREATE TRIGGER deliveries_no_update BEFORE UPDATE ON deliveries BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END;
|
||||
CREATE TRIGGER deliveries_no_delete BEFORE DELETE ON deliveries BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END;
|
||||
CREATE TRIGGER meta_no_replace BEFORE INSERT ON meta WHEN EXISTS (SELECT 1 FROM meta WHERE key = NEW.key) BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END;
|
||||
CREATE TRIGGER events_no_replace BEFORE INSERT ON events WHEN EXISTS (SELECT 1 FROM events WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'events is append-only'); END;
|
||||
CREATE TRIGGER role_claims_no_replace BEFORE INSERT ON role_claims WHEN EXISTS (SELECT 1 FROM role_claims WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END;
|
||||
CREATE TRIGGER decisions_no_replace BEFORE INSERT ON decisions WHEN EXISTS (SELECT 1 FROM decisions WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END;
|
||||
CREATE TRIGGER decision_events_no_replace BEFORE INSERT ON decision_events WHEN EXISTS (SELECT 1 FROM decision_events WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END;
|
||||
CREATE TRIGGER messages_no_replace BEFORE INSERT ON messages WHEN EXISTS (SELECT 1 FROM messages WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END;
|
||||
CREATE TRIGGER deliveries_no_replace BEFORE INSERT ON deliveries WHEN EXISTS (SELECT 1 FROM deliveries WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END;
|
||||
CREATE TRIGGER task_snapshots_no_update BEFORE UPDATE ON task_snapshots BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END;
|
||||
CREATE TRIGGER task_snapshots_no_delete BEFORE DELETE ON task_snapshots BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END;
|
||||
CREATE TRIGGER task_snapshots_no_replace BEFORE INSERT ON task_snapshots WHEN EXISTS (SELECT 1 FROM task_snapshots WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END;
|
||||
CREATE TRIGGER decisions_blocking_needs_task BEFORE INSERT ON decisions
|
||||
WHEN NEW.blocking = 1 AND NEW.task_ref IS NULL
|
||||
BEGIN SELECT RAISE(ABORT, 'a blocking decision cites the task it blocks'); END;
|
||||
CREATE TRIGGER events_launch_by_human BEFORE INSERT ON events
|
||||
WHEN NEW.kind IN ('launch.revoked','launch.restored') AND (NEW.actor_role IS NOT NULL OR NEW.actor_run IS NOT NULL)
|
||||
BEGIN SELECT RAISE(ABORT, 'only the human revokes or restores launching'); END;
|
||||
CREATE TRIGGER events_credential_body BEFORE INSERT ON events
|
||||
WHEN NEW.kind GLOB 'credential.*' AND (
|
||||
json_extract(NEW.body, '$.service') IS NULL OR json_extract(NEW.body, '$.service') NOT IN ('gitea','vikunja')
|
||||
OR json_extract(NEW.body, '$.instance') IS NULL)
|
||||
BEGIN SELECT RAISE(ABORT, 'credential events name a service and a role instance'); END;
|
||||
CREATE TRIGGER events_task_missing_body BEFORE INSERT ON events
|
||||
WHEN NEW.kind = 'task.missing' AND (
|
||||
json_extract(NEW.body, '$.reason') IS NULL OR json_extract(NEW.body, '$.reason') NOT IN ('moved','not-found','no-access')
|
||||
OR (json_extract(NEW.body, '$.reason') = 'moved' AND json_type(NEW.body, '$.project') IS NOT 'integer'))
|
||||
BEGIN SELECT RAISE(ABORT, 'task.missing carries a reason, and the new project when moved'); END;
|
||||
CREATE TRIGGER events_task_subject BEFORE INSERT ON events
|
||||
WHEN NEW.kind GLOB 'task.*' AND (NEW.subject IS NULL OR NOT (NEW.subject GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(NEW.subject, 9) GLOB '*[^0-9/]*' AND NOT substr(NEW.subject, 9) GLOB '*/*/*'))
|
||||
BEGIN SELECT RAISE(ABORT, 'a task event names its task in subject'); END;
|
||||
CREATE TRIGGER events_task_created_body BEFORE INSERT ON events
|
||||
WHEN NEW.kind = 'task.created' AND (
|
||||
json_type(NEW.body, '$.request') IS NOT 'text'
|
||||
OR NOT EXISTS (SELECT 1 FROM events WHERE id = json_extract(NEW.body, '$.request')
|
||||
AND kind = 'human.input' AND business = NEW.business)
|
||||
OR json_type(NEW.body, '$.requirement') IS NOT 'text'
|
||||
OR NOT json_extract(NEW.body, '$.requirement') GLOB 'REQ-[A-Z]*-[1-9]*'
|
||||
OR json_extract(NEW.body, '$.requirement') GLOB 'REQ-*[^A-Z0-9-]*')
|
||||
BEGIN SELECT RAISE(ABORT, 'task.created cites the human.input that asked for it and a requirement id'); END;
|
||||
CREATE VIEW launch_state AS
|
||||
SELECT business, CASE kind WHEN 'launch.revoked' THEN 'revoked' ELSE 'allowed' END AS state, at
|
||||
FROM events e WHERE kind IN ('launch.revoked','launch.restored')
|
||||
AND seq = (SELECT max(seq) FROM events WHERE business = e.business AND kind IN ('launch.revoked','launch.restored'));
|
||||
CREATE VIEW task_external_changes AS
|
||||
SELECT p.business, p.task_ref, p.seq, p.via, p.updated, p.digest, ls.digest AS self_digest
|
||||
FROM task_snapshots p
|
||||
LEFT JOIN task_snapshots ls ON ls.seq = (SELECT max(seq) FROM task_snapshots
|
||||
WHERE business = p.business AND task_ref = p.task_ref AND source = 'self')
|
||||
WHERE p.source = 'poll'
|
||||
AND p.seq = (SELECT max(seq) FROM task_snapshots WHERE business = p.business AND task_ref = p.task_ref)
|
||||
AND (ls.seq IS NULL OR (p.updated >= ls.updated AND p.read_at > ls.at AND p.digest <> ls.digest));
|
||||
CREATE VIEW task_current AS
|
||||
SELECT s.seq, s.at, s.business, s.task_ref, s.updated, s.etag, s.digest, s.fields, s.source, s.via, s.read_at, s.role, s.run
|
||||
FROM task_snapshots s
|
||||
WHERE s.seq = (SELECT max(c.seq) FROM task_snapshots c
|
||||
WHERE c.business = s.business AND c.task_ref = s.task_ref
|
||||
AND (c.source = 'self'
|
||||
OR c.read_at > coalesce((SELECT ls.at FROM task_snapshots ls
|
||||
WHERE ls.business = c.business AND ls.task_ref = c.task_ref AND ls.source = 'self'
|
||||
ORDER BY ls.seq DESC LIMIT 1), '')));
|
||||
CREATE VIEW tasks_open AS
|
||||
SELECT s.business, s.task_ref, json_extract(s.fields, '$.bucket') AS bucket, s.seq
|
||||
FROM task_current s
|
||||
WHERE json_type(s.fields, '$.gone') IS NULL
|
||||
AND json_extract(s.fields, '$.done') = 0;
|
||||
CREATE VIEW urgent_inbox AS
|
||||
SELECT d.id, d.business, d.task_ref, d.question, d.at
|
||||
FROM decisions d
|
||||
WHERE d.class = 'gated' AND d.blocking = 1
|
||||
AND NOT EXISTS (SELECT 1 FROM decision_events x WHERE x.decision = d.id AND x.op IN ('resolved','withdrawn','expired'));
|
||||
@@ -516,7 +516,7 @@ test('v3b prototype refusals, views and append-only mutations', (t) => {
|
||||
|
||||
const count = (type) => db.prepare('SELECT count(*) n FROM sqlite_master WHERE type = ?').get(type).n;
|
||||
for (const label of refused) assert.ok(tested.has(label), label);
|
||||
assert.equal(count('table') - 1, 8);
|
||||
assert.equal(count('trigger'), 35);
|
||||
assert.equal(count('table') - 1, 9);
|
||||
assert.equal(count('trigger'), 39);
|
||||
assert.equal(count('view'), 5);
|
||||
});
|
||||
|
||||
@@ -427,3 +427,22 @@ run use a scratch container on 127.0.0.1, never a shared instance.
|
||||
|
||||
`deploy/vikunja/` has the compose file for installs without their own
|
||||
Vikunja. See its README.
|
||||
|
||||
### Durable comment cursor (row 49)
|
||||
|
||||
The bus stores each business/task's highest observed comment ID separately from
|
||||
`task_current` and its digest. Adapter startup restores those IDs before its
|
||||
first reconcile. The first read of each task's comment list after startup catches
|
||||
outage comments, even outside the normal `updated` window. A reconcile also
|
||||
checks comment lists when the expanded count changes. Bot IDs advance the cursor
|
||||
but do not produce external-comment counts. The cursor and event commit together;
|
||||
in-memory ID, count and timestamp caches advance only after a successful commit.
|
||||
A crash after commit reloads the durable cursor and does not repeat the event.
|
||||
|
||||
No cursor row is written until a new ID is observed. Empty tasks implicitly have
|
||||
cursor zero; initial timestamps use the database's creation/migration epoch minus
|
||||
a 60-second skew margin. This can duplicate comments at the first migration if
|
||||
the tracker clock is ahead, and clocks more than a minute behind remain outside
|
||||
the guarantee. Once a cursor exists, timestamps do not govern freshness.
|
||||
See the bus README for the explicit offline schema migration and Mos's separate
|
||||
approved downtime/re-pin procedure. No development seat migrates the live store.
|
||||
|
||||
@@ -68,6 +68,7 @@ export function tasksAdapter({
|
||||
const call = client({ baseUrl: s.baseUrl, fetch });
|
||||
const use = (instance) => (method, path, opts) =>
|
||||
credentials.use(`${name}/${instance}`, 'vikunja', (token) => call(token, method, path, opts));
|
||||
const comments = broker.taskView(name, 'comments');
|
||||
const ctx = {
|
||||
...s,
|
||||
broker,
|
||||
@@ -86,7 +87,8 @@ export function tasksAdapter({
|
||||
pending: 0,
|
||||
lastStart: null,
|
||||
lastReconcile: null,
|
||||
lastComment: new Map(),
|
||||
lastComment: new Map(comments.cursors.map((c) => [c.task_id, c.last_comment])),
|
||||
commentEpoch: Date.parse(comments.epoch),
|
||||
commentsAt: new Map(),
|
||||
commentCount: new Map(),
|
||||
files: Object.fromEntries(Object.entries(files(businesses[name])).map(([i, f]) => [i, { file: f, stamp: stamp(f) }])),
|
||||
|
||||
+18
-14
@@ -79,12 +79,14 @@ export async function walk(ctx, filter, query = {}, project = ctx.project) {
|
||||
const changed = (before, after) =>
|
||||
Object.keys(after).filter((k) => JSON.stringify(before?.[k]) !== JSON.stringify(after[k]));
|
||||
// Comments by anyone but this business's bots, newer than the last one seen. The first look at a
|
||||
// task counts only comments created inside the window, so a restart doesn't replay old ones.
|
||||
// task uses the database epoch minus the skew margin; later reads use durable IDs.
|
||||
// `created` is to the second, so the bound is too.
|
||||
async function foreignComments(ctx, id, sinceMs) {
|
||||
async function foreignComments(ctx, id, out) {
|
||||
const r = await all(ctx.syncCall, null, `/tasks/${id}/comments`, {}, ctx.per);
|
||||
if (!r.ok) throw new BusError(refusal(r.r));
|
||||
const since = Math.floor(sinceMs / 1000) * 1000;
|
||||
// No cursor yet: allow one minute of tracker clock skew around creation/migration.
|
||||
// Once an ID is durable, tracker timestamps never affect freshness.
|
||||
const since = Math.floor((ctx.commentEpoch - WINDOW_MS) / 1000) * 1000;
|
||||
const last = ctx.lastComment.get(id);
|
||||
let n = 0,
|
||||
top = last ?? 0;
|
||||
@@ -94,7 +96,7 @@ async function foreignComments(ctx, id, sinceMs) {
|
||||
const fresh = last === undefined ? Date.parse(c.created) >= since : c.id > last;
|
||||
if (fresh && !ctx.bots.has(c.author.id)) n++;
|
||||
}
|
||||
ctx.lastComment.set(id, top);
|
||||
if (top > (last ?? 0)) out.commentCursors.push({ task_id: id, last_comment: top });
|
||||
return n;
|
||||
}
|
||||
// One task read. Records nothing that a newer self snapshot already covers, and nothing that
|
||||
@@ -163,7 +165,9 @@ async function missing(ctx, current, refs, out) {
|
||||
}
|
||||
}
|
||||
const record = (ctx, out) => {
|
||||
if (out.snapshots.length || out.events.length) ctx.broker.recordTask({ business: ctx.business, ...out });
|
||||
if (out.snapshots.length || out.events.length || out.commentCursors.length)
|
||||
ctx.broker.recordTask({ business: ctx.business, ...out });
|
||||
for (const c of out.commentCursors) ctx.lastComment.set(c.task_id, c.last_comment);
|
||||
};
|
||||
// A child that answers 403 or 404 may have left the tree after the last reconcile: no longer
|
||||
// shared, moved under a project the bot can't read, or deleted. A 404 can also be the child's kanban
|
||||
@@ -191,17 +195,16 @@ export async function tick(ctx) {
|
||||
}
|
||||
}
|
||||
const current = new Map(ctx.broker.taskView(ctx.business, 'current').map((x) => [x.task_ref, x]));
|
||||
const out = { snapshots: [], events: [] };
|
||||
const out = { snapshots: [], events: [], commentCursors: [] };
|
||||
// Refs some read this tick placed. Keyed by ref, not id: a task moved between two projects of the
|
||||
// tree shows on the new one's board under the same id, and the old ref still needs its tombstone.
|
||||
const placed = new Set();
|
||||
for (const { project, view, open, hits } of reads) {
|
||||
const comments = new Map();
|
||||
for (const [id, t] of hits) {
|
||||
const key = normal(t.updated);
|
||||
if (ctx.commentsAt.get(id) === key) continue;
|
||||
comments.set(id, await foreignComments(ctx, id, sinceMs));
|
||||
ctx.commentsAt.set(id, key);
|
||||
if (ctx.commentsAt.get(id) === normal(t.updated)) continue;
|
||||
comments.set(id, await foreignComments(ctx, id, out));
|
||||
// Cache advances only after the database transaction commits.
|
||||
}
|
||||
for (const [id, { task, bucket }] of open) {
|
||||
const hit = hits.get(id);
|
||||
@@ -224,6 +227,7 @@ export async function tick(ctx) {
|
||||
.filter((r) => !placed.has(r));
|
||||
await missing(ctx, current, gone, out);
|
||||
record(ctx, out);
|
||||
for (const { hits } of reads) for (const [id, t] of hits) ctx.commentsAt.set(id, normal(t.updated));
|
||||
ctx.lastStart = startMs;
|
||||
return { snapshots: out.snapshots.length, events: out.events.length };
|
||||
}
|
||||
@@ -234,13 +238,12 @@ export async function tick(ctx) {
|
||||
export async function reconcile(ctx) {
|
||||
const startMs = ctx.clock();
|
||||
const readAt = iso(startMs);
|
||||
const sinceMs = ctx.lastReconcile ?? startMs;
|
||||
if (ctx.descendants) await discover(ctx);
|
||||
const reads = [];
|
||||
for (const [project, view] of ctx.tree)
|
||||
reads.push({ project, view, open: await openTasks(ctx, project, view), list: await walk(ctx, '', { expand: 'comment_count' }, project) });
|
||||
const current = new Map(ctx.broker.taskView(ctx.business, 'current').map((x) => [x.task_ref, x]));
|
||||
const out = { snapshots: [], events: [] };
|
||||
const out = { snapshots: [], events: [], commentCursors: [] };
|
||||
const listed = new Set();
|
||||
for (const { project, view, open, list } of reads)
|
||||
for (const t of list) {
|
||||
@@ -251,8 +254,7 @@ export async function reconcile(ctx) {
|
||||
const count = t.comment_count;
|
||||
if (count !== undefined && !Number.isSafeInteger(count)) shape();
|
||||
const before = ctx.commentCount.get(t.id);
|
||||
if (count !== undefined) ctx.commentCount.set(t.id, count);
|
||||
if (before !== undefined && count !== before) comments = await foreignComments(ctx, t.id, sinceMs);
|
||||
if (before === undefined || count !== before) comments = await foreignComments(ctx, t.id, out);
|
||||
consider(ctx, current, project, { task: t, bucket, via: 'reconcile', readAt, comments }, out);
|
||||
}
|
||||
const gone = [...current.values()]
|
||||
@@ -260,6 +262,8 @@ export async function reconcile(ctx) {
|
||||
.map((x) => x.task_ref);
|
||||
await missing(ctx, current, gone, out);
|
||||
record(ctx, out);
|
||||
for (const { list } of reads)
|
||||
for (const t of list) if (t.comment_count !== undefined) ctx.commentCount.set(t.id, t.comment_count);
|
||||
ctx.lastReconcile = startMs;
|
||||
ctx.lastStart ??= startMs;
|
||||
return { snapshots: out.snapshots.length, events: out.events.length };
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { world } from './world.mjs';
|
||||
|
||||
async function fixture(t) {
|
||||
let now = Date.now();
|
||||
const w = await world(t, { fake: { now: () => now }, adapter: { clock: () => now } });
|
||||
await w.adapter.start('demo');
|
||||
const owner = await w.ui('POST', `/projects/${w.project}/tasks`, { title: 'watched' });
|
||||
assert.equal(owner.status, 201);
|
||||
await w.adapter.reconcile('demo');
|
||||
assert.equal(w.store.get('SELECT count(*) n FROM task_comment_cursors').n, 0);
|
||||
const events = () => w.events('task.changed.external').filter(e => e.body.comments);
|
||||
return { w, events, advance: () => { now += 3 * 3600000; } };
|
||||
}
|
||||
|
||||
test('stopped bus resumes an empty task cursor after a long outage, exactly once', async t => {
|
||||
const { w, events, advance } = await fixture(t);
|
||||
await w.stop();
|
||||
advance();
|
||||
assert.equal((await w.ui('POST', '/tasks/1/comments', { comment: 'while bus stopped' })).status, 201);
|
||||
advance();
|
||||
await w.restart();
|
||||
assert.equal(events().length, 1);
|
||||
assert.equal(events()[0].body.comments, 1);
|
||||
await w.adapter.tick('demo');
|
||||
await w.adapter.reconcile('demo');
|
||||
assert.equal(events().length, 1);
|
||||
});
|
||||
|
||||
test('persisted IDs avoid replay, exclude bots and bound rows to observed new comments', async t => {
|
||||
const { w, events, advance } = await fixture(t);
|
||||
await w.ui('POST', '/tasks/1/comments', { comment: 'before restart' });
|
||||
await w.adapter.reconcile('demo');
|
||||
assert.equal(events().length, 1);
|
||||
const rows = () => w.store.get('SELECT count(*) n FROM task_comment_cursors').n;
|
||||
assert.equal(rows(), 1);
|
||||
await w.adapter.reconcile('demo');
|
||||
await w.adapter.tick('demo');
|
||||
assert.equal(rows(), 1);
|
||||
await w.stop();
|
||||
advance();
|
||||
const r = await w.fake.fetch('http://vikunja.test/api/v2/tasks/1/comments', {
|
||||
method: 'POST', headers: { authorization: 'Bearer ' + w.tokens.coder, 'content-type': 'application/json' },
|
||||
body: JSON.stringify({ comment: 'bot while stopped' }),
|
||||
});
|
||||
assert.equal(r.status, 201);
|
||||
await w.ui('POST', '/tasks/1/comments', { comment: 'human while stopped' });
|
||||
advance();
|
||||
await w.restart();
|
||||
assert.equal(events().length, 2);
|
||||
assert.equal(events()[1].body.comments, 1);
|
||||
assert.equal(rows(), 2);
|
||||
await w.stop();
|
||||
await w.restart();
|
||||
await w.adapter.tick('demo');
|
||||
assert.equal(events().length, 2);
|
||||
assert.equal(rows(), 2);
|
||||
});
|
||||
|
||||
test('commit followed by crash before cache advance neither loses nor duplicates a comment', async t => {
|
||||
const { w, events } = await fixture(t);
|
||||
await w.ui('POST', '/tasks/1/comments', { comment: 'committed before crash' });
|
||||
const record = w.broker.recordTask.bind(w.broker);
|
||||
w.broker.recordTask = args => { record(args); throw Error('simulated crash after commit'); };
|
||||
await assert.rejects(w.adapter.reconcile('demo'), /simulated crash/);
|
||||
assert.equal(events().length, 1);
|
||||
await w.stop();
|
||||
await w.restart();
|
||||
assert.equal(events().length, 1);
|
||||
await w.adapter.tick('demo');
|
||||
assert.equal(events().length, 1);
|
||||
});
|
||||
|
||||
test('failed event insert rolls back cursor and a retry still emits the comment once', async t => {
|
||||
const { w, events } = await fixture(t);
|
||||
await w.ui('POST', '/tasks/1/comments', { comment: 'retry me' });
|
||||
w.store.run("CREATE TRIGGER fixture_fail BEFORE INSERT ON events WHEN NEW.kind='task.changed.external' BEGIN SELECT RAISE(ABORT,'fixture failure'); END");
|
||||
await assert.rejects(w.adapter.reconcile('demo'), /fixture failure/);
|
||||
w.store.run('DROP TRIGGER fixture_fail');
|
||||
await w.adapter.reconcile('demo');
|
||||
assert.equal(events().length, 1);
|
||||
assert.equal(events()[0].body.comments, 1);
|
||||
});
|
||||
|
||||
// Removing the epoch margin loses a tracker comment thirty seconds behind host time.
|
||||
test('initial cursor accepts a comment within the stated one-minute clock skew margin', async t => {
|
||||
let trackerNow = Date.now() - 30000;
|
||||
const skewed = await world(t, { fake: { now: () => trackerNow } });
|
||||
await skewed.adapter.start('demo');
|
||||
await skewed.ui('POST', `/projects/${skewed.project}/tasks`, { title: 'clock skew' });
|
||||
await skewed.adapter.reconcile('demo');
|
||||
await skewed.ui('POST', '/tasks/1/comments', { comment: 'thirty seconds behind' });
|
||||
await skewed.adapter.reconcile('demo');
|
||||
const found = skewed.events('task.changed.external').filter(e => e.body.comments);
|
||||
assert.equal(found.length, 1);
|
||||
assert.equal(found[0].body.comments, 1);
|
||||
await skewed.adapter.reconcile('demo');
|
||||
assert.equal(skewed.events('task.changed.external').filter(e => e.body.comments).length, 1);
|
||||
});
|
||||
@@ -104,8 +104,8 @@ export async function world(t, edit = {}) {
|
||||
for (const [r, x] of Object.entries(roles)) if (x.credentials) references['demo/' + r] = x.credentials;
|
||||
if (businesses.demo.tracker.sync.credentials) references['demo/@sync'] = businesses.demo.tracker.sync.credentials;
|
||||
const credentials = new Credentials({ references, env, dataRoot: root });
|
||||
const store = new Store(root);
|
||||
const broker = new Broker({ store, businesses, secretCheck: (v) => credentials.assertClean(v) });
|
||||
let store = new Store(root);
|
||||
let broker = new Broker({ store, businesses, secretCheck: (v) => credentials.assertClean(v) });
|
||||
const log = [];
|
||||
const factory = tasksAdapter({
|
||||
trackers: { demo: { baseUrl: 'http://vikunja.test', project, ...edit.tracker } },
|
||||
@@ -114,7 +114,7 @@ export async function world(t, edit = {}) {
|
||||
autostart: false,
|
||||
...edit.adapter,
|
||||
});
|
||||
const adapter = await factory({ broker, credentials, businesses });
|
||||
let adapter = await factory({ broker, credentials, businesses });
|
||||
t.after(async () => {
|
||||
await adapter.close();
|
||||
store.close();
|
||||
@@ -145,6 +145,17 @@ export async function world(t, edit = {}) {
|
||||
store.all('SELECT * FROM task_snapshots WHERE task_ref=? ORDER BY seq', taskRef).map((s) => ({ ...s, fields: JSON.parse(s.fields) }));
|
||||
const call = (cap, verb, args) => broker.requestTask(cap, { verb, args }, adapter.handle);
|
||||
return {
|
||||
async stop() {
|
||||
await adapter.close();
|
||||
store.close();
|
||||
},
|
||||
async restart() {
|
||||
store = new Store(root);
|
||||
broker = new Broker({ store, businesses, secretCheck: (v) => credentials.assertClean(v) });
|
||||
adapter = await factory({ broker, credentials, businesses });
|
||||
await adapter.start('demo');
|
||||
if (adapter.status()[0].state !== 'ready') throw Error(adapter.status()[0].refused);
|
||||
},
|
||||
root,
|
||||
fake,
|
||||
owner,
|
||||
@@ -155,9 +166,9 @@ export async function world(t, edit = {}) {
|
||||
tokens,
|
||||
credentials,
|
||||
businesses,
|
||||
store,
|
||||
broker,
|
||||
adapter,
|
||||
get store() { return store; },
|
||||
get broker() { return broker; },
|
||||
get adapter() { return adapter; },
|
||||
log,
|
||||
caps,
|
||||
agent,
|
||||
|
||||
@@ -10,18 +10,14 @@
|
||||
// "refuse<TAB>reason<TAB>detail" line and exit 2. A missing or unsafe
|
||||
// database or business file exits 3.
|
||||
//
|
||||
// Kept in step with packages/bus/src/store.mjs, which this imports from:
|
||||
// the schema_version literal '3b' below is a copy of the Store's. A schema
|
||||
// bump changes it there, and until it changes here every read refuses
|
||||
// bus-schema.
|
||||
// Schema version and digest follow the Store exports; a schema bump has one authority.
|
||||
import { DatabaseSync } from 'node:sqlite';
|
||||
import { readFileSync, lstatSync } from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
import { pathToFileURL } from 'node:url';
|
||||
import { createHash } from 'node:crypto';
|
||||
import { schemaDigest, SCHEMA_SHA256 } from '../../packages/bus/src/store.mjs';
|
||||
import { schemaDigest, SCHEMA_SHA256, SCHEMA_VERSION } from '../../packages/bus/src/store.mjs';
|
||||
|
||||
const SCHEMA_VERSION = '3b';
|
||||
const MERGE = 'git.merge.protected';
|
||||
const WORK_REF = /^work\/vikunja-([1-9][0-9]*)-([1-9][0-9]*)$/;
|
||||
const ANY_WORK_REF = /\bwork\/[A-Za-z0-9._/-]*[A-Za-z0-9_]/g;
|
||||
|
||||
Reference in New Issue
Block a user