148 lines
6.3 KiB
Go
148 lines
6.3 KiB
Go
package spool
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
"encoding/hex"
|
|
"fmt"
|
|
)
|
|
|
|
type migration struct {
|
|
version uint32
|
|
sql string
|
|
}
|
|
|
|
var migrations = []migration{
|
|
{version: 1, sql: schemaV1},
|
|
{version: 2, sql: schemaV2},
|
|
{version: 3, sql: schemaV3},
|
|
}
|
|
|
|
func applyMigrations(ctx context.Context, db *sql.DB) error {
|
|
if _, err := db.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations (
|
|
version INTEGER PRIMARY KEY CHECK(version > 0), checksum TEXT NOT NULL, applied_at INTEGER NOT NULL
|
|
) STRICT`); err != nil {
|
|
return fmt.Errorf("create client spool migration table: %w", err)
|
|
}
|
|
for _, current := range migrations {
|
|
checksumBytes := sha256.Sum256([]byte(current.sql))
|
|
checksum := hex.EncodeToString(checksumBytes[:])
|
|
var stored string
|
|
err := db.QueryRowContext(ctx, `SELECT checksum FROM schema_migrations WHERE version = ?`, current.version).Scan(&stored)
|
|
if err == nil {
|
|
if stored != checksum {
|
|
return fmt.Errorf("client spool migration %d checksum mismatch", current.version)
|
|
}
|
|
continue
|
|
}
|
|
if err != sql.ErrNoRows {
|
|
return err
|
|
}
|
|
tx, err := db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err = tx.ExecContext(ctx, current.sql); err == nil {
|
|
_, err = tx.ExecContext(ctx, `INSERT INTO schema_migrations(version, checksum, applied_at) VALUES (?, ?, unixepoch())`, current.version, checksum)
|
|
}
|
|
if err != nil {
|
|
_ = tx.Rollback()
|
|
return fmt.Errorf("apply client spool migration %d: %w", current.version, err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return fmt.Errorf("commit client spool migration %d: %w", current.version, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
const schemaV1 = `
|
|
CREATE TABLE commands (
|
|
issue_uuid BLOB PRIMARY KEY CHECK(length(issue_uuid) = 16),
|
|
immutable_sha256 BLOB NOT NULL CHECK(length(immutable_sha256) = 32),
|
|
command_revision INTEGER NOT NULL CHECK(command_revision > 0),
|
|
phase INTEGER NOT NULL CHECK(phase BETWEEN 1 AND 11),
|
|
terminal INTEGER NOT NULL DEFAULT 0 CHECK(terminal IN (0, 1)),
|
|
output_charged_bytes INTEGER NOT NULL DEFAULT 0 CHECK(output_charged_bytes >= 0),
|
|
base_charged_bytes INTEGER NOT NULL CHECK(base_charged_bytes > 0),
|
|
total_charged_bytes INTEGER NOT NULL CHECK(total_charged_bytes >= 0),
|
|
closeout_remaining_bytes INTEGER NOT NULL CHECK(closeout_remaining_bytes >= 0),
|
|
next_local_ordinal INTEGER NOT NULL DEFAULT 1 CHECK(next_local_ordinal > 0),
|
|
next_event_seq INTEGER NOT NULL DEFAULT 1 CHECK(next_event_seq > 0),
|
|
last_server_ack INTEGER NOT NULL DEFAULT 0 CHECK(last_server_ack >= 0),
|
|
accepted_at INTEGER NOT NULL
|
|
) STRICT;
|
|
CREATE TABLE events (
|
|
issue_uuid BLOB NOT NULL REFERENCES commands(issue_uuid) ON DELETE CASCADE,
|
|
local_ordinal INTEGER NOT NULL CHECK(local_ordinal > 0),
|
|
event_seq INTEGER CHECK(event_seq IS NULL OR event_seq > 0),
|
|
event_kind INTEGER NOT NULL CHECK(event_kind > 0),
|
|
compression INTEGER NOT NULL CHECK(compression IN (1, 2)),
|
|
raw_bytes INTEGER NOT NULL CHECK(raw_bytes >= 0),
|
|
charged_bytes INTEGER NOT NULL CHECK(charged_bytes > 0),
|
|
output INTEGER NOT NULL CHECK(output IN (0, 1)),
|
|
payload BLOB NOT NULL,
|
|
payload_sha256 BLOB NOT NULL CHECK(length(payload_sha256) = 32),
|
|
created_at INTEGER NOT NULL,
|
|
PRIMARY KEY(issue_uuid, local_ordinal),
|
|
UNIQUE(issue_uuid, event_seq)
|
|
) STRICT, WITHOUT ROWID;
|
|
CREATE INDEX events_send_window ON events(issue_uuid, event_seq, local_ordinal);
|
|
CREATE TABLE scripts (
|
|
issue_uuid BLOB PRIMARY KEY REFERENCES commands(issue_uuid) ON DELETE CASCADE CHECK(length(issue_uuid) = 16),
|
|
declared_raw_bytes INTEGER NOT NULL CHECK(declared_raw_bytes >= 0),
|
|
declared_sha256 BLOB NOT NULL CHECK(length(declared_sha256) = 32),
|
|
received_raw_bytes INTEGER NOT NULL DEFAULT 0 CHECK(received_raw_bytes >= 0),
|
|
stored_bytes INTEGER NOT NULL CHECK(stored_bytes >= 0),
|
|
compression INTEGER NOT NULL CHECK(compression = 2),
|
|
stored_data BLOB NOT NULL,
|
|
charged_bytes INTEGER NOT NULL CHECK(charged_bytes > 0),
|
|
committed INTEGER NOT NULL DEFAULT 0 CHECK(committed IN (0, 1)),
|
|
CHECK(received_raw_bytes <= declared_raw_bytes),
|
|
CHECK((committed = 0) OR received_raw_bytes = declared_raw_bytes)
|
|
) STRICT, WITHOUT ROWID;
|
|
CREATE TABLE spool_counters (
|
|
singleton INTEGER PRIMARY KEY CHECK(singleton = 1),
|
|
client_total_charged_bytes INTEGER NOT NULL CHECK(client_total_charged_bytes >= 0),
|
|
charge_version INTEGER NOT NULL CHECK(charge_version = 1)
|
|
) STRICT;
|
|
INSERT INTO spool_counters(singleton, client_total_charged_bytes, charge_version) VALUES (1, 0, 1);
|
|
CREATE TABLE command_tombstones (
|
|
issue_uuid BLOB PRIMARY KEY CHECK(length(issue_uuid) = 16),
|
|
immutable_sha256 BLOB NOT NULL CHECK(length(immutable_sha256) = 32),
|
|
command_revision INTEGER NOT NULL CHECK(command_revision > 0),
|
|
terminal_lifecycle INTEGER NOT NULL CHECK(terminal_lifecycle BETWEEN 5 AND 11),
|
|
acknowledged_at INTEGER NOT NULL
|
|
) STRICT;
|
|
CREATE INDEX tombstones_fifo ON command_tombstones(acknowledged_at, issue_uuid);
|
|
`
|
|
|
|
// schemaV2 adds the immutable execution specification retained with every
|
|
// accepted dispatch. Keeping it in its own table lets old commands created by
|
|
// early development builds (which have no specification) remain readable and
|
|
// makes the payload's quota charge independently auditable.
|
|
const schemaV2 = `
|
|
CREATE TABLE command_specs (
|
|
issue_uuid BLOB PRIMARY KEY REFERENCES commands(issue_uuid) ON DELETE CASCADE CHECK(length(issue_uuid) = 16),
|
|
raw_bytes INTEGER NOT NULL CHECK(raw_bytes > 0),
|
|
stored_bytes INTEGER NOT NULL CHECK(stored_bytes > 0),
|
|
compression INTEGER NOT NULL CHECK(compression = 2),
|
|
payload BLOB NOT NULL,
|
|
payload_sha256 BLOB NOT NULL CHECK(length(payload_sha256) = 32),
|
|
charged_bytes INTEGER NOT NULL CHECK(charged_bytes > 0)
|
|
) STRICT, WITHOUT ROWID;
|
|
`
|
|
|
|
// schemaV3 adds a small durable launch barrier to every accepted command. A
|
|
// value of 2 means launch authorization may have crossed the OS boundary; on
|
|
// restart the client must interrupt that command instead of redispatching it.
|
|
// Keeping these fields on commands makes the barrier part of the existing
|
|
// command-owned quota/accounting row and lets terminal cleanup remove it with
|
|
// the command.
|
|
const schemaV3 = `
|
|
ALTER TABLE commands ADD COLUMN launch_phase INTEGER NOT NULL DEFAULT 0 CHECK(launch_phase BETWEEN 0 AND 2);
|
|
ALTER TABLE commands ADD COLUMN launch_context TEXT NOT NULL DEFAULT '';
|
|
ALTER TABLE commands ADD COLUMN launch_pid INTEGER NOT NULL DEFAULT 0 CHECK(launch_pid >= 0);
|
|
`
|