amadan

carlos / platform Public

console: pending-change store — payload-pinned quorum with 7-day expiry

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 0fe8e88520b0639078209a33787a9a310377e2a0 parent 8d37d61
3 files changed, +525 −1
  • internal/console/pending.go +335 −0
  • internal/console/pending_test.go +185 −0
  • internal/console/store.go +5 −1
diff --git a/internal/console/pending.go b/internal/console/pending.go
new file mode 100644
index 000000000..66dc5b326
--- /dev/null
+++ b/internal/console/pending.go
@@ -0,0 +1,335 @@
+package console
+
+// Pending changes: multi-party quorum for the actions release-pipelines-v2
+// gates (spec §4) — promote, policy, fasttrack. A change is proposed once,
+// with its payload pinned at propose time (canonical JSON — Task 8 re-checks
+// nothing here, it only ever executes the payload this row already holds),
+// and members confirm it until a quorum is met.
+//
+// The load-bearing rule (spec §4, controller ruling): a BEARER TOKEN NEVER
+// COUNTS TOWARD A PASSKEY QUORUM. A console/step-up proposal has already
+// proved presence, so its proposer is recorded as the first confirmation in
+// the same transaction that creates the row. A CLI bearer-token proposal has
+// proved nothing of the kind, so it starts at zero confirmations — even a
+// quorum of 1 needs a SEPARATE, passkey-backed confirmation before it is
+// met, including one made by the same identity that proposed it over the
+// CLI.
+
+import (
+ "crypto/rand"
+ "database/sql"
+ "encoding/hex"
+ "fmt"
+ "time"
+)
+
+// PendingChangeTTL bounds how long a proposed change waits for quorum before
+// it can no longer be confirmed or executed.
+const PendingChangeTTL = 7 * 24 * time.Hour
+
+// PendingChange is one proposed, quorum-gated action.
+type PendingChange struct {
+ ID string
+ AccountID int64
+ App string
+ Kind string // "promote" | "policy" | "fasttrack"
+ Payload []byte // canonical JSON, pinned at propose time
+ Quorum int
+ Proposer string
+ CreatedAt time.Time
+ ExpiresAt time.Time
+ Confirms []string // addresses, proposer included when passkey-backed
+}
+
+// migratePendingChanges adds this feature's two tables. Additive and
+// idempotent, and kept here rather than in store.go's list so the feature is
+// one file to read and one file to remove (the authstore.go / stepup.go
+// precedent).
+func (s *Store) migratePendingChanges() error {
+ for _, stmt := range []string{
+ `CREATE TABLE IF NOT EXISTS pending_changes (
+ id TEXT PRIMARY KEY,
+ account_id INTEGER NOT NULL,
+ app TEXT NOT NULL,
+ kind TEXT NOT NULL,
+ payload TEXT NOT NULL,
+ quorum INTEGER NOT NULL,
+ proposer TEXT NOT NULL,
+ created_at INTEGER NOT NULL,
+ expires_at INTEGER NOT NULL,
+ executed_at INTEGER NOT NULL DEFAULT 0
+);`,
+ `CREATE TABLE IF NOT EXISTS pending_confirms (
+ change_id TEXT NOT NULL,
+ address TEXT NOT NULL,
+ confirmed_at INTEGER NOT NULL,
+ PRIMARY KEY (change_id, address)
+);`,
+ } {
+ if _, err := s.db.Exec(stmt); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// newChangeID mints a pending-change id: 16 bytes of crypto/rand, hex —
+// the same shape as the keymail envelope id (keymail.go), unguessable and
+// URL-safe with no encoding fuss.
+func newChangeID() (string, error) {
+ b := make([]byte, 16)
+ if _, err := rand.Read(b); err != nil {
+ return "", err
+ }
+ return hex.EncodeToString(b), nil
+}
+
+// ProposeChange inserts a change; proposerConfirmed records the proposer's
+// confirmation row (true only for console/step-up proposals — spec §4: a
+// bearer token never counts toward a passkey quorum). One transaction: a
+// console/step-up proposal's row and its first confirmation must never be
+// separable by a crash, or a reader could observe a change credited to
+// nobody despite the proposer having just proved presence for it.
+func (s *Store) ProposeChange(accountID int64, app, kind string, payload []byte, quorum int, proposer string, proposerConfirmed bool, now time.Time) (*PendingChange, error) {
+ id, err := newChangeID()
+ if err != nil {
+ return nil, err
+ }
+ created := now.Unix()
+ expires := now.Add(PendingChangeTTL).Unix()
+ tx, err := s.db.Begin()
+ if err != nil {
+ return nil, err
+ }
+ defer tx.Rollback()
+ if _, err := tx.Exec(`INSERT INTO pending_changes
+(id, account_id, app, kind, payload, quorum, proposer, created_at, expires_at)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+ id, accountID, app, kind, payload, quorum, proposer, created, expires); err != nil {
+ return nil, err
+ }
+ var confirms []string
+ if proposerConfirmed {
+ if _, err := tx.Exec(`INSERT INTO pending_confirms (change_id, address, confirmed_at) VALUES (?, ?, ?)`,
+ id, proposer, created); err != nil {
+ return nil, err
+ }
+ confirms = []string{proposer}
+ }
+ if err := tx.Commit(); err != nil {
+ return nil, err
+ }
+ return &PendingChange{
+ ID: id,
+ AccountID: accountID,
+ App: app,
+ Kind: kind,
+ Payload: payload,
+ Quorum: quorum,
+ Proposer: proposer,
+ CreatedAt: time.Unix(created, 0).UTC(),
+ ExpiresAt: time.Unix(expires, 0).UTC(),
+ Confirms: confirms,
+ }, nil
+}
+
+// ConfirmChange records one member's confirmation (idempotent per address —
+// INSERT OR IGNORE, then a recount, so confirming twice never inflates the
+// tally) and reports whether quorum is now met, including when it was
+// already met before this call. Expired (now >= ExpiresAt) or executed
+// changes refuse outright: neither state accepts a new confirmation. One
+// transaction throughout, so the read that decides "expired"/"executed" and
+// the write that records the confirmation can never straddle a concurrent
+// MarkExecuted or a second confirmation.
+func (s *Store) ConfirmChange(id, address string, now time.Time) (*PendingChange, bool, error) {
+ tx, err := s.db.Begin()
+ if err != nil {
+ return nil, false, err
+ }
+ defer tx.Rollback()
+ pc, err := loadPendingChange(tx, id)
+ if err != nil {
+ return nil, false, err
+ }
+ if pc == nil {
+ return nil, false, fmt.Errorf("console: no such pending change %q", id)
+ }
+ if now.Unix() >= pc.ExpiresAt.Unix() {
+ return nil, false, fmt.Errorf("console: pending change %q has expired", id)
+ }
+ if !executedAtZero(tx, id) {
+ return nil, false, fmt.Errorf("console: pending change %q has already been executed", id)
+ }
+ if _, err := tx.Exec(`INSERT OR IGNORE INTO pending_confirms (change_id, address, confirmed_at) VALUES (?, ?, ?)`,
+ id, address, now.Unix()); err != nil {
+ return nil, false, err
+ }
+ confirms, err := confirmsFor(tx, id)
+ if err != nil {
+ return nil, false, err
+ }
+ pc.Confirms = confirms
+ if err := tx.Commit(); err != nil {
+ return nil, false, err
+ }
+ return pc, len(confirms) >= pc.Quorum, nil
+}
+
+// PendingChangesFor lists an account/app's unexpired, unexecuted changes.
+func (s *Store) PendingChangesFor(accountID int64, app string, now time.Time) ([]PendingChange, error) {
+ tx, err := s.db.Begin()
+ if err != nil {
+ return nil, err
+ }
+ defer tx.Rollback()
+ rows, err := tx.Query(`SELECT id, account_id, app, kind, payload, quorum, proposer, created_at, expires_at
+FROM pending_changes
+WHERE account_id = ? AND app = ? AND executed_at = 0 AND expires_at > ?
+ORDER BY created_at, id`, accountID, app, now.Unix())
+ if err != nil {
+ return nil, err
+ }
+ var ids []string
+ var out []PendingChange
+ for rows.Next() {
+ var pc PendingChange
+ var created, expires int64
+ if err := rows.Scan(&pc.ID, &pc.AccountID, &pc.App, &pc.Kind, &pc.Payload, &pc.Quorum, &pc.Proposer,
+ &created, &expires); err != nil {
+ rows.Close()
+ return nil, err
+ }
+ pc.CreatedAt = time.Unix(created, 0).UTC()
+ pc.ExpiresAt = time.Unix(expires, 0).UTC()
+ out = append(out, pc)
+ ids = append(ids, pc.ID)
+ }
+ rows.Close()
+ if err := rows.Err(); err != nil {
+ return nil, err
+ }
+ for i, id := range ids {
+ confirms, err := confirmsFor(tx, id)
+ if err != nil {
+ return nil, err
+ }
+ out[i].Confirms = confirms
+ }
+ if err := tx.Commit(); err != nil {
+ return nil, err
+ }
+ return out, nil
+}
+
+// PendingChange fetches one change, excluding executed and expired rows —
+// both read as (nil, nil), the same "not found" a caller gets for an id that
+// was never proposed at all.
+func (s *Store) PendingChange(id string, now time.Time) (*PendingChange, error) {
+ tx, err := s.db.Begin()
+ if err != nil {
+ return nil, err
+ }
+ defer tx.Rollback()
+ pc, err := loadPendingChange(tx, id)
+ if err != nil {
+ return nil, err
+ }
+ if pc == nil {
+ return nil, nil
+ }
+ if now.Unix() >= pc.ExpiresAt.Unix() {
+ return nil, nil
+ }
+ if !executedAtZero(tx, id) {
+ return nil, nil
+ }
+ confirms, err := confirmsFor(tx, id)
+ if err != nil {
+ return nil, err
+ }
+ pc.Confirms = confirms
+ if err := tx.Commit(); err != nil {
+ return nil, err
+ }
+ return pc, nil
+}
+
+// MarkExecuted stamps a change as carried out, so it stops appearing as
+// pending and can no longer be confirmed.
+func (s *Store) MarkExecuted(id string, now time.Time) error {
+ _, err := s.db.Exec(`UPDATE pending_changes SET executed_at = ? WHERE id = ? AND executed_at = 0`,
+ now.Unix(), id)
+ return err
+}
+
+// DeleteChange removes a change outright — reject/cancel. Its confirmations
+// go with it: a deleted change is gone, not merely emptied.
+func (s *Store) DeleteChange(id string) error {
+ tx, err := s.db.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+ if _, err := tx.Exec(`DELETE FROM pending_confirms WHERE change_id = ?`, id); err != nil {
+ return err
+ }
+ if _, err := tx.Exec(`DELETE FROM pending_changes WHERE id = ?`, id); err != nil {
+ return err
+ }
+ return tx.Commit()
+}
+
+// ---- helpers ----------------------------------------------------------------
+
+// loadPendingChange reads a change row by id, unfiltered by expiry or
+// execution — the callers above apply those rules themselves, since
+// ConfirmChange needs to distinguish "expired" from "executed" for its error
+// message and PendingChange/PendingChangesFor need to treat both as absence.
+// Confirms is left nil; callers fill it via confirmsFor.
+func loadPendingChange(tx *sql.Tx, id string) (*PendingChange, error) {
+ var pc PendingChange
+ var created, expires int64
+ err := tx.QueryRow(`SELECT id, account_id, app, kind, payload, quorum, proposer, created_at, expires_at
+FROM pending_changes WHERE id = ?`, id).
+ Scan(&pc.ID, &pc.AccountID, &pc.App, &pc.Kind, &pc.Payload, &pc.Quorum, &pc.Proposer, &created, &expires)
+ if err == sql.ErrNoRows {
+ return nil, nil
+ }
+ if err != nil {
+ return nil, err
+ }
+ pc.CreatedAt = time.Unix(created, 0).UTC()
+ pc.ExpiresAt = time.Unix(expires, 0).UTC()
+ return &pc, nil
+}
+
+// executedAtZero reports whether a change's executed_at is still 0 (i.e. it
+// has not been executed). A missing row reads as "not executed" — the
+// caller has already confirmed the row exists via loadPendingChange before
+// this is asked.
+func executedAtZero(tx *sql.Tx, id string) bool {
+ var executedAt int64
+ if err := tx.QueryRow(`SELECT executed_at FROM pending_changes WHERE id = ?`, id).Scan(&executedAt); err != nil {
+ return true // let the caller's own row-existence check own "not found"
+ }
+ return executedAt == 0
+}
+
+// confirmsFor returns a change's confirming addresses, sorted by when they
+// confirmed.
+func confirmsFor(tx *sql.Tx, id string) ([]string, error) {
+ rows, err := tx.Query(`SELECT address FROM pending_confirms WHERE change_id = ? ORDER BY confirmed_at, address`, id)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []string
+ for rows.Next() {
+ var a string
+ if err := rows.Scan(&a); err != nil {
+ return nil, err
+ }
+ out = append(out, a)
+ }
+ return out, rows.Err()
+}
diff --git a/internal/console/pending_test.go b/internal/console/pending_test.go
new file mode 100644
index 000000000..5c5ddcb4d
--- /dev/null
+++ b/internal/console/pending_test.go
@@ -0,0 +1,185 @@
+package console
+
+import (
+ "reflect"
+ "sort"
+ "strings"
+ "testing"
+ "time"
+)
+
+func TestPendingChangeQuorum(t *testing.T) {
+ s := openTestStore(t)
+ now := time.Now()
+ accountID := testAccountID()
+
+ // Console-proposed (proposerConfirmed=true): the proposer's confirmation
+ // is recorded atomically with the change, but that alone must not meet a
+ // quorum of 2.
+ pc, err := s.ProposeChange(accountID, "kass", "promote", []byte(`{"v":1}`), 2, "paul@keymail.dev", true, now)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if pc.ID == "" {
+ t.Fatal("ProposeChange: empty ID")
+ }
+ if !reflect.DeepEqual(pc.Confirms, []string{"paul@keymail.dev"}) {
+ t.Fatalf("ProposeChange Confirms = %v, want [paul@keymail.dev]", pc.Confirms)
+ }
+ if pc.Quorum != 2 || pc.AccountID != accountID || pc.App != "kass" || pc.Kind != "promote" {
+ t.Fatalf("ProposeChange fields wrong: %+v", pc)
+ }
+ if string(pc.Payload) != `{"v":1}` {
+ t.Fatalf("ProposeChange Payload = %s", pc.Payload)
+ }
+ if !pc.ExpiresAt.Equal(pc.CreatedAt.Add(PendingChangeTTL)) {
+ t.Fatalf("ExpiresAt = %v, want CreatedAt+%v = %v", pc.ExpiresAt, PendingChangeTTL, pc.CreatedAt.Add(PendingChangeTTL))
+ }
+
+ // Re-confirming as the SAME proposer is idempotent: still one confirm,
+ // not met.
+ got, met, err := s.ConfirmChange(pc.ID, "paul@keymail.dev", now)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if met {
+ t.Fatal("ConfirmChange: met=true after idempotent re-confirm by proposer, want false")
+ }
+ if !reflect.DeepEqual(got.Confirms, []string{"paul@keymail.dev"}) {
+ t.Fatalf("Confirms after re-confirm = %v, want [paul@keymail.dev]", got.Confirms)
+ }
+
+ // A second, distinct member's confirmation reaches quorum.
+ got, met, err = s.ConfirmChange(pc.ID, "bob@keymail.dev", now)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !met {
+ t.Fatal("ConfirmChange: met=false after quorum reached, want true")
+ }
+ wantConfirms := []string{"bob@keymail.dev", "paul@keymail.dev"}
+ gotConfirms := append([]string(nil), got.Confirms...)
+ sort.Strings(gotConfirms)
+ if !reflect.DeepEqual(gotConfirms, wantConfirms) {
+ t.Fatalf("Confirms at quorum = %v, want %v", gotConfirms, wantConfirms)
+ }
+
+ // Already-met quorum still reports met on a further read.
+ fetched, err := s.PendingChange(pc.ID, now)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if fetched == nil {
+ t.Fatal("PendingChange returned nil for a live change")
+ }
+ if len(fetched.Confirms) != 2 {
+ t.Fatalf("PendingChange Confirms = %v, want 2 entries", fetched.Confirms)
+ }
+}
+
+func TestPendingChangeCLIProposalStartsAtZero(t *testing.T) {
+ s := openTestStore(t)
+ now := time.Now()
+ accountID := testAccountID()
+
+ // CLI bearer-token proposal (proposerConfirmed=false): a bearer token
+ // must never count toward a passkey quorum, so this starts at zero
+ // confirmations even though quorum is only 1.
+ pc, err := s.ProposeChange(accountID, "aria", "policy", []byte(`{"v":2}`), 1, "cli@keymail.dev", false, now)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(pc.Confirms) != 0 {
+ t.Fatalf("ProposeChange(proposerConfirmed=false) Confirms = %v, want empty", pc.Confirms)
+ }
+
+ list, err := s.PendingChangesFor(accountID, "aria", now)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(list) != 1 || len(list[0].Confirms) != 0 {
+ t.Fatalf("PendingChangesFor = %+v, want one change with zero confirms", list)
+ }
+
+ // One console confirmation — even by the proposer themself — meets
+ // quorum of 1.
+ got, met, err := s.ConfirmChange(pc.ID, "cli@keymail.dev", now)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !met {
+ t.Fatal("ConfirmChange: met=false after single confirmation on quorum-1 CLI proposal, want true")
+ }
+ if !reflect.DeepEqual(got.Confirms, []string{"cli@keymail.dev"}) {
+ t.Fatalf("Confirms = %v, want [cli@keymail.dev]", got.Confirms)
+ }
+}
+
+func TestPendingChangeExpiryAndExecution(t *testing.T) {
+ s := openTestStore(t)
+ now := time.Now()
+ accountID := testAccountID()
+
+ // --- expiry ---
+ pc, err := s.ProposeChange(accountID, "kass", "fasttrack", []byte(`{"v":3}`), 2, "paul@keymail.dev", true, now)
+ if err != nil {
+ t.Fatal(err)
+ }
+ past := pc.ExpiresAt.Add(time.Second) // now >= ExpiresAt
+ _, _, err = s.ConfirmChange(pc.ID, "bob@keymail.dev", past)
+ if err == nil || !strings.Contains(err.Error(), "expired") {
+ t.Fatalf("ConfirmChange past expiry: err = %v, want an error mentioning 'expired'", err)
+ }
+
+ list, err := s.PendingChangesFor(accountID, "kass", past)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(list) != 0 {
+ t.Fatalf("PendingChangesFor at/after expiry = %v, want empty", list)
+ }
+ if got, err := s.PendingChange(pc.ID, past); err != nil || got != nil {
+ t.Fatalf("PendingChange at/after expiry = %v, %v, want nil, nil", got, err)
+ }
+
+ // --- execution ---
+ pc2, err := s.ProposeChange(accountID, "kass", "promote", []byte(`{"v":4}`), 1, "paul@keymail.dev", true, now)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := s.MarkExecuted(pc2.ID, now); err != nil {
+ t.Fatal(err)
+ }
+ _, _, err = s.ConfirmChange(pc2.ID, "bob@keymail.dev", now)
+ if err == nil {
+ t.Fatal("ConfirmChange on an executed change: want error, got nil")
+ }
+
+ // pc (above) is still genuinely pending at real `now` — only `past` made
+ // it look expired — so it legitimately remains in this account/app's
+ // listing; what must be true is that the newly-executed pc2 does not.
+ list2, err := s.PendingChangesFor(accountID, "kass", now)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, c := range list2 {
+ if c.ID == pc2.ID {
+ t.Fatalf("PendingChangesFor after execution still lists executed change %s: %+v", pc2.ID, list2)
+ }
+ }
+ if got, err := s.PendingChange(pc2.ID, now); err != nil || got != nil {
+ t.Fatalf("PendingChange after execution = %v, %v, want nil, nil", got, err)
+ }
+
+ // --- delete ---
+ pc3, err := s.ProposeChange(accountID, "kass", "promote", []byte(`{"v":5}`), 1, "paul@keymail.dev", true, now)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := s.DeleteChange(pc3.ID); err != nil {
+ t.Fatal(err)
+ }
+ if got, err := s.PendingChange(pc3.ID, now); err != nil || got != nil {
+ t.Fatalf("PendingChange after DeleteChange = %v, %v, want nil, nil", got, err)
+ }
+}
diff --git a/internal/console/store.go b/internal/console/store.go
index 8d950b2e4..4b3c9b340 100644
--- a/internal/console/store.go
+++ b/internal/console/store.go
@@ -237,7 +237,11 @@ func (s *Store) migrate() error {
return err
}
// Passkey step-up's own tables (stepup.go), for the same reason.
- return s.migrateStepUp()
+ if err := s.migrateStepUp(); err != nil {
+ return err
+ }
+ // Pending-change quorum's own tables (pending.go), for the same reason.
+ return s.migratePendingChanges()
}
// migrateAppsKey moves the apps table from PRIMARY KEY(app) — a GLOBAL name