M-6: today's rows carried over once, before the manager does anything else (issue 1409)

At its first start the manager reads the six tables of its database, read only, and writes them into its kept data and its grants into the vault at generation 1, then carried.postgres; until then only its read verbs answer. A second start never opens the database, and a stopped run is done again whole without doubling a log entry. The store's write code and its postgres test go: nothing writes the database again. Builds: design 39 bus-and-switch §1, §8, §11, §12, §13, A3, B14 Claude-Session: https://claude.ai/code/session_01NcbgfHb9d4T9nq5Q6WqdiJ
2026-11-11 23:28:14 +01:00
parent 2b604e0dad
commit 9cc62818f4
7 changed files with 877 additions and 341 deletions
@@ -0,0 +1,437 @@
+package main
+
+// The carry-over (design 39 bus-and-switch §8, §11 to §13, A3, B14): at its first start the manager reads every row of
+// today's database, the module's own and read only, and writes it into its kept data on the bus and its grants into
+// the vault, then the key `carried.postgres`. Until that key exists it serves its read verbs alone and does nothing
+// else. With the key present it never reads the database again.
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+)
+
+// CarriedKey is the key of the state `carried` the carry-over writes at its end (§8).
+const CarriedKey = "postgres"
+
+// NotCarried is the answer of every acting verb and tool until the carry-over has ended (§11).
+const NotCarried = "the carry-over has not ended"
+
+// usageKeptFor is how far back the carry-over reads usage: the log `usage` keeps no older reading (§6, §12).
+const usageKeptFor = 35 * 24 * time.Hour
+
+// The tables of today's database, in the order the carry-over reads them (§11, §12).
+var CarriedTables = []string{"licence", "binding", "lease", "offered", "usage", "audit"}
+
+// LicenceRow is a row of the table `licence`: the licence and its grant at rest, sealed with `grant.key`.
+type LicenceRow struct {
+	Licence
+	Sealed string
+}
+
+// OfferedRow is a row of the table `offered`.
+type OfferedRow struct {
+	Fingerprint, Node, Account string
+	Outcome                    Outcome
+	Why                        string
+	At                         time.Time
+}
+
+// UsageTableRow is a row of the table `usage`: the reading and the vendor's answer whole.
+type UsageTableRow struct {
+	Licence string
+	At      int64 // milliseconds
+	Reading UsageReading
+	Raw     map[string]any
+}
+
+// AuditRow is a row of the table `audit`.
+type AuditRow struct {
+	At     time.Time
+	What   string
+	Detail map[string]any
+}
+
+// CarrySource reads the six tables of today's database, each oldest first; a table that does not exist is no rows.
+type CarrySource interface {
+	Licences(ctx context.Context) ([]LicenceRow, error)
+	Bindings(ctx context.Context) ([]Binding, error)
+	Leases(ctx context.Context) (int, error)
+	Offered(ctx context.Context) ([]OfferedRow, error)
+	Usage(ctx context.Context, since time.Time) ([]UsageTableRow, error)
+	Audit(ctx context.Context) ([]AuditRow, error)
+}
+
+// CarriedTable is one table in `carried.postgres`: the rows read and the entries written for them.
+type CarriedTable struct {
+	Read    int `json:"read"`
+	Written int `json:"written"`
+}
+
+// CarriedRecord is the value of `carried.postgres` (§8): when, and for each table the rows read and entries written.
+type CarriedRecord struct {
+	At     string                  `json:"at"`
+	Tables map[string]CarriedTable `json:"tables"`
+}
+
+// Carry is the carry-over's gate: closed until `carried.postgres` exists, and the failure of its last run.
+type Carry struct {
+	State keptState // the state `carried`
+	// Source opens today's database; called only while the key is absent.
+	Source func(ctx context.Context) (CarrySource, error)
+
+	mu      sync.Mutex
+	running bool
+	failure string
+	done    chan struct{}
+	once    sync.Once
+}
+
+func (c *Carry) init() {
+	c.mu.Lock()
+	if c.done == nil {
+		c.done = make(chan struct{})
+	}
+	c.mu.Unlock()
+}
+
+// Ended is whether the carry-over has ended: the key `carried.postgres` exists.
+func (c *Carry) Ended() bool {
+	if c == nil {
+		return true // a manager built with no carry-over to wait on: the tests' own
+	}
+	c.init()
+	select {
+	case <-c.done:
+		return true
+	default:
+	}
+	if e, err := c.State.Get(CarriedKey); err == nil && e != nil && len(e.Value) > 0 {
+		c.end()
+		return true
+	}
+	return false
+}
+
+func (c *Carry) end() { c.once.Do(func() { close(c.done) }) }
+
+// Wait is closed once the carry-over has ended.
+func (c *Carry) Wait() <-chan struct{} {
+	c.init()
+	return c.done
+}
+
+// Acts answers an acting verb or tool: nothing until the carry-over has ended (§11).
+func (c *Carry) Acts() error {
+	if c.Ended() {
+		return nil
+	}
+	return errors.New(NotCarried)
+}
+
+// Failure is what the last run that failed names, empty when none failed.
+func (c *Carry) Failure() string {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.failure
+}
+
+// Run carries over every row when `carried.postgres` is absent, and says nothing else happened when it is present
+// (§13). A run that fails names its table and error, and leaves the key absent, so the next is done again whole.
+func (c *Carry) Run(ctx context.Context, m *Manager, s *BusStore) (*CarriedRecord, error) {
+	if c.Ended() {
+		return nil, nil
+	}
+	c.mu.Lock()
+	if c.running {
+		c.mu.Unlock()
+		return nil, errors.New("a carry-over is running")
+	}
+	c.running = true
+	c.mu.Unlock()
+	rec, err := c.run(ctx, m, s)
+	c.mu.Lock()
+	c.running = false
+	if err != nil {
+		c.failure = err.Error()
+	} else {
+		c.failure = ""
+	}
+	c.mu.Unlock()
+	if err != nil {
+		m.Log("the carry-over failed: %v", err)
+		return nil, err
+	}
+	c.end()
+	m.Log("carried over today's database: %v", rec.Tables)
+	return rec, nil
+}
+
+func (c *Carry) run(ctx context.Context, m *Manager, s *BusStore) (*CarriedRecord, error) {
+	src, err := c.Source(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("the database cannot be opened: %w", err)
+	}
+	if closer, ok := src.(interface{ Close() }); ok {
+		defer closer.Close()
+	}
+	co := carryOver{m: m, s: s, src: src, now: m.Now(), tables: map[string]CarriedTable{}}
+	if err := co.all(ctx); err != nil {
+		return nil, err
+	}
+	rec := CarriedRecord{At: co.now.UTC().Format(time.RFC3339), Tables: co.tables}
+	detail := map[string]any{"why": "the first start of the manager on the bus", "by": Itself, "tables": rec.Tables}
+	if err := s.Audit(ctx, "carried", detail); err != nil {
+		return nil, fmt.Errorf("carried: %w", err)
+	}
+	if _, err := c.State.Put(CarriedKey, rec); err != nil {
+		return nil, fmt.Errorf("carried: the key %s was not written: %w", CarriedKey, err)
+	}
+	return &rec, nil
+}
+
+// carryOver is one run of the carry-over.
+type carryOver struct {
+	m      *Manager
+	s      *BusStore
+	src    CarrySource
+	now    time.Time
+	tables map[string]CarriedTable
+	audit  []AuditRow
+}
+
+func (co *carryOver) all(ctx context.Context) error {
+	// The audit is read first: a licence's adoptedFrom comes from it (B14).
+	audit, err := co.src.Audit(ctx)
+	if err != nil {
+		return fmt.Errorf("audit: %w", err)
+	}
+	co.audit = audit
+	for _, table := range CarriedTables {
+		var t CarriedTable
+		var err error
+		switch table {
+		case "licence":
+			t, err = co.licences(ctx)
+		case "binding":
+			t, err = co.bindings(ctx)
+		case "lease":
+			t.Read, err = co.src.Leases(ctx) // a lease lives two minutes: none taken before the start still holds
+		case "offered":
+			t, err = co.offered(ctx)
+		case "usage":
+			t, err = co.usage(ctx)
+		case "audit":
+			t, err = co.audits()
+		}
+		if err != nil {
+			return fmt.Errorf("%s: %w", table, err)
+		}
+		co.tables[table] = t
+	}
+	return nil
+}
+
+// adoptedFrom is the node the newest adoption of a licence names in the audit (B14).
+func (co *carryOver) adoptedFrom(name string) string {
+	from := ""
+	for _, r := range co.audit {
+		if r.What != "adopted" || r.Detail["licence"] != name {
+			continue
+		}
+		if n, _ := r.Detail["node"].(string); n != "" {
+			from = n
+		} else if f, _ := r.Detail["from"].(string); f != "" {
+			from = f
+		}
+	}
+	return from
+}
+
+// licences writes a key of `licences` per row, its grant opened with `grant.key` and handed to the vault at
+// generation 1, the grant first (§12, §15).
+func (co *carryOver) licences(ctx context.Context) (CarriedTable, error) {
+	rows, err := co.src.Licences(ctx)
+	if err != nil {
+		return CarriedTable{}, err
+	}
+	t := CarriedTable{Read: len(rows)}
+	for _, r := range rows {
+		l := r.Licence
+		l.Generation, l.Usage = 0, nil
+		l.AdoptedFrom = co.adoptedFrom(l.Name)
+		if r.Sealed != "" {
+			if co.m.Crypt == nil {
+				return t, fmt.Errorf("the grant of %s cannot be opened: no grant.key", l.Name)
+			}
+			plain, err := co.m.Crypt.Open(r.Sealed)
+			if err != nil {
+				return t, fmt.Errorf("the grant of %s: %w", l.Name, err)
+			}
+			grant, access := plain, plain
+			if l.Kind == "subscription" {
+				var g FullGrant
+				if err := json.Unmarshal([]byte(plain), &g); err != nil {
+					return t, fmt.Errorf("the grant of %s is not a grant", l.Name)
+				}
+				access = accessOf(g)
+			}
+			if err := co.m.handOver(ctx, &l, grant, access); err != nil {
+				return t, err
+			}
+		}
+		if err := co.s.SaveLicence(ctx, l); err != nil {
+			return t, err
+		}
+		t.Written++
+	}
+	return t, nil
+}
+
+// bindings writes a key of `bindings` per row, its generation kept, why "carried over", by the manager, since the
+// carry-over's time (§12, B14); a key a run before this one wrote for the same row keeps its since, so it is put again
+// with the same value (§13).
+func (co *carryOver) bindings(ctx context.Context) (CarriedTable, error) {
+	rows, err := co.src.Bindings(ctx)
+	if err != nil {
+		return CarriedTable{}, err
+	}
+	kinds := map[string]string{}
+	if all, err := co.s.Licences(ctx); err == nil {
+		for _, l := range all {
+			kinds[l.Name] = l.Kind
+		}
+	} else {
+		return CarriedTable{}, err
+	}
+	t := CarriedTable{Read: len(rows)}
+	for _, r := range rows {
+		v := BindingState{Licence: r.Licence, Kind: kinds[r.Licence], Generation: r.Generation,
+			Since: co.now.UTC().Format(time.RFC3339), Why: "carried over", By: Itself}
+		var held BindingState
+		if ok, err := co.s.get(co.s.BindingsState, r.Consumer, &held); err != nil {
+			return t, err
+		} else if ok && held.Licence == v.Licence && held.Generation == v.Generation && held.Why == v.Why {
+			v.Since = held.Since
+		}
+		if _, err := co.s.BindingsState.Put(r.Consumer, v); err != nil {
+			return t, err
+		}
+		t.Written++
+	}
+	return t, nil
+}
+
+// offered writes a key of `offered` per row, its lifetime counted from the row's `at`; a row past its lifetime is
+// gone already and written nowhere (§5, §12).
+func (co *carryOver) offered(ctx context.Context) (CarriedTable, error) {
+	rows, err := co.src.Offered(ctx)
+	if err != nil {
+		return CarriedTable{}, err
+	}
+	t := CarriedTable{Read: len(rows)}
+	for _, r := range rows {
+		left := offeredFor - co.now.Sub(r.At)
+		if left <= 0 {
+			continue
+		}
+		v := offeredValue{Node: r.Node, Account: r.Account, Outcome: r.Outcome, Why: r.Why,
+			At: r.At.UTC().Format(time.RFC3339)}
+		if err := co.s.putOffered(r.Fingerprint, v, left); err != nil {
+			return t, err
+		}
+		t.Written++
+	}
+	return t, nil
+}
+
+// lastAt is the `at` of a log's last entry for a subject, zero when it has none: rows at or before it are in the log
+// already (§13).
+func lastAt(l keptLog, key string) (time.Time, error) {
+	e, err := l.Last(key)
+	if err != nil || e == nil {
+		return time.Time{}, err
+	}
+	var v struct {
+		At string `json:"at"`
+	}
+	if json.Unmarshal(e.Value, &v) != nil {
+		return time.Time{}, nil
+	}
+	at, _ := time.Parse(time.RFC3339Nano, v.At)
+	return at, nil
+}
+
+// usage appends one entry of the log `usage` per row of the last 35 days, oldest first, its `at` kept (§12).
+func (co *carryOver) usage(ctx context.Context) (CarriedTable, error) {
+	rows, err := co.src.Usage(ctx, co.now.Add(-usageKeptFor))
+	if err != nil {
+		return CarriedTable{}, err
+	}
+	t := CarriedTable{Read: len(rows)}
+	last := map[string]time.Time{}
+	for _, r := range rows {
+		key := KeyFor(r.Licence)
+		before, seen := last[key]
+		if !seen {
+			if before, err = lastAt(co.s.UsageLog, key); err != nil {
+				return t, err
+			}
+			last[key] = before
+		}
+		at := time.UnixMilli(r.At).UTC()
+		if !at.After(before) {
+			t.Written++ // in the log already, from a run before this one
+			continue
+		}
+		if _, err := co.s.UsageLog.Append(key, UsageEntry{At: at.Format(time.RFC3339Nano), Reading: r.Reading,
+			Raw: r.Raw}); err != nil {
+			return t, err
+		}
+		t.Written++
+	}
+	return t, nil
+}
+
+// audits appends one entry of the log `audit` per row, oldest first, its `at` kept, under the subject its `what`
+// names, `drift` for any other (§7, §12).
+func (co *carryOver) audits() (CarriedTable, error) {
+	t := CarriedTable{Read: len(co.audit)}
+	last := map[string]time.Time{}
+	for _, r := range co.audit {
+		kind := "drift"
+		for _, k := range AuditKinds {
+			if k == r.What {
+				kind = k
+			}
+		}
+		before, seen := last[kind]
+		if !seen {
+			var err error
+			if before, err = lastAt(co.s.AuditLog, kind); err != nil {
+				return t, err
+			}
+			last[kind] = before
+		}
+		if !r.At.After(before) {
+			t.Written++ // in the log already, from a run before this one
+			continue
+		}
+		entry := map[string]any{}
+		for k, v := range r.Detail {
+			entry[k] = v
+		}
+		entry["at"] = r.At.UTC().Format(time.RFC3339Nano)
+		if kind != r.What {
+			entry["what"] = r.What
+		}
+		if _, err := co.s.AuditLog.Append(kind, entry); err != nil {
+			return t, fmt.Errorf("the audit did not take the carried %s entry: %w", r.What, err)
+		}
+		t.Written++
+	}
+	return t, nil
+}
@@ -0,0 +1,276 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"os"
+	"regexp"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+// memSource is today's database as the carry-over reads it, from rows given; failAt makes one table's read fail.
+type memSource struct {
+	licences []LicenceRow
+	bindings []Binding
+	leases   int
+	offered  []OfferedRow
+	usage    []UsageTableRow
+	audit    []AuditRow
+	failAt   string
+}
+
+func (s *memSource) fail(table string) error {
+	if s.failAt == table {
+		return errors.New("connection reset by peer")
+	}
+	return nil
+}
+func (s *memSource) Licences(context.Context) ([]LicenceRow, error) {
+	return s.licences, s.fail("licence")
+}
+func (s *memSource) Bindings(context.Context) ([]Binding, error) {
+	return s.bindings, s.fail("binding")
+}
+func (s *memSource) Leases(context.Context) (int, error) { return s.leases, s.fail("lease") }
+func (s *memSource) Offered(context.Context) ([]OfferedRow, error) {
+	return s.offered, s.fail("offered")
+}
+func (s *memSource) Audit(context.Context) ([]AuditRow, error) { return s.audit, s.fail("audit") }
+func (s *memSource) Usage(_ context.Context, since time.Time) ([]UsageTableRow, error) {
+	var out []UsageTableRow
+	for _, u := range s.usage {
+		if u.At >= since.UnixMilli() {
+			out = append(out, u)
+		}
+	}
+	return out, s.fail("usage")
+}
+
+// todaysRows is a copy of the shape of today's rows: a subscription and an API key, two bindings, a lease, two offers,
+// three readings (one older than 35 days) and four audit rows (one of a kind the log does not name).
+func todaysRows(t *testing.T, mm *miniMesh) *memSource {
+	t.Helper()
+	grant, _ := json.Marshal(FullGrant{AccessToken: "at-old", RefreshToken: "rt-old", ExpiresAt: 7})
+	pct := 42.0
+	return &memSource{
+		licences: []LicenceRow{
+			{Licence: Licence{Name: licence1, Kind: "subscription", AccountUUID: "acct-1", Email: licence1,
+				RefreshFingerprint: Fingerprint("rt-old"), AccessExpiresAt: 7, AdoptedAt: 1}, Sealed: sealAtRest(t, mm.m.Crypt, string(grant))},
+			{Licence: Licence{Name: "api", Kind: "api-key", AdoptedAt: 2}, Sealed: sealAtRest(t, mm.m.Crypt, "sk-ant-api03-x")},
+		},
+		bindings: []Binding{{Consumer: "laptop", Licence: licence1, Generation: 1013}, {Consumer: "server", Licence: "api", Generation: 1016}},
+		leases:   1,
+		offered: []OfferedRow{{Fingerprint: "sha256:aa", Node: "laptop", Account: "acct-1", Outcome: Adopted, Why: "refreshed", At: t0.Add(-24 * time.Hour)},
+			{Fingerprint: "sha256:bb", Node: "server", Outcome: Dead, Why: "400", At: t0.Add(-48 * time.Hour)}},
+		usage: []UsageTableRow{
+			{Licence: licence1, At: t0.Add(-40 * 24 * time.Hour).UnixMilli(), Reading: UsageReading{SessionPct: &pct}},
+			{Licence: licence1, At: t0.Add(-2 * time.Hour).UnixMilli(), Reading: UsageReading{SessionPct: &pct}, Raw: map[string]any{"five_hour": 1.0}},
+			{Licence: licence1, At: t0.Add(-1 * time.Hour).UnixMilli(), Reading: UsageReading{SessionPct: &pct}},
+		},
+		audit: []AuditRow{
+			{At: t0.Add(-72 * time.Hour), What: "adopted", Detail: map[string]any{"licence": licence1, "node": "laptop"}},
+			{At: t0.Add(-71 * time.Hour), What: "bound", Detail: map[string]any{"consumer": "laptop"}},
+			{At: t0.Add(-70 * time.Hour), What: "switched", Detail: map[string]any{"consumer": "server"}},
+			{At: t0.Add(-69 * time.Hour), What: "grant-moved", Detail: map[string]any{"licence": "api"}},
+		},
+	}
+}
+
+func carryFor(mm *miniMesh, carried *memState, src CarrySource) *Carry {
+	return &Carry{State: carried, Source: func(context.Context) (CarrySource, error) { return src, nil }}
+}
+
+func readCarried(t *testing.T, carried *memState) CarriedRecord {
+	t.Helper()
+	var rec CarriedRecord
+	e, _ := carried.Get(CarriedKey)
+	if e == nil {
+		t.Fatal("carried.postgres is not written")
+	}
+	_ = json.Unmarshal(e.Value, &rec)
+	return rec
+}
+
+// design 39 bus-and-switch §11/AC1, §12/AC1: every row is carried over and the key written; the counts in
+// carried.postgres equal the rows read for every table but lease, and the vault answers each licence's grant.
+func TestEveryRowIsCarriedOver(t *testing.T) {
+	mm := newMesh(t)
+	carried := newMemState(func() time.Time { return mm.now })
+	src := todaysRows(t, mm)
+	c := carryFor(mm, carried, src)
+	if _, err := c.Run(context.Background(), mm.m, mm.store); err != nil {
+		t.Fatal(err)
+	}
+	rec := readCarried(t, carried)
+	want := map[string]int{"licence": 2, "binding": 2, "lease": 1, "offered": 2, "usage": 2, "audit": 4}
+	for table, n := range want {
+		got := rec.Tables[table]
+		if got.Read != n || (table != "lease" && got.Written != got.Read) {
+			t.Errorf("%s: read %d written %d, want %d read and as many written", table, got.Read, got.Written, n)
+		}
+	}
+	if g, err := mm.grants.Get(context.Background(), KeyFor(licence1), PartGrant); err != nil || !strings.Contains(g, "rt-old") {
+		t.Fatalf("the vault answers the grant %q, %v", g, err)
+	}
+	if a, _ := mm.grants.Get(context.Background(), KeyFor(licence1), PartAccess); strings.Contains(a, "rt-old") || !strings.Contains(a, "at-old") {
+		t.Fatalf("the access secret is %q", a)
+	}
+	if a, _ := mm.grants.Get(context.Background(), KeyFor("api"), PartAccess); a != "sk-ant-api03-x" {
+		t.Fatalf("the API key's access is %q", a)
+	}
+	l, _ := mm.store.Licence(context.Background(), licence1)
+	if l == nil || l.Generation != 1 || l.AdoptedFrom != "laptop" || l.AccountUUID != "acct-1" || l.AccessExpiresAt != 7 {
+		t.Fatalf("the licence is %+v", l)
+	}
+	if b := mm.bound("laptop"); b.Generation != 1013 || b.Why != "carried over" || b.By != Itself || b.Kind != "subscription" || b.Since == "" {
+		t.Fatalf("laptop's binding %+v", b)
+	}
+	if u := mm.offered.until[OfferedKey("sha256:aa")]; !u.Equal(t0.Add(-24 * time.Hour).Add(offeredFor)) {
+		t.Fatalf("the offer lives until %v", u)
+	}
+	if got := mm.audit.under("drift"); len(got) != 1 || got[0]["licence"] != "api" {
+		t.Fatalf("drift holds %v", got)
+	}
+	if got := mm.audit.under("adopted"); len(got) != 1 || got[0]["at"] != t0.Add(-72*time.Hour).Format(time.RFC3339Nano) {
+		t.Fatalf("adopted holds %v", got)
+	}
+	if got := mm.usage.under(KeyFor(licence1)); len(got) != 2 || got[0]["raw"] == nil {
+		t.Fatalf("usage holds %v", got)
+	}
+	if len(mm.audit.under("carried")) != 1 || !c.Ended() {
+		t.Fatal("the carry-over did not say it carried, or did not end")
+	}
+}
+
+// design 39 bus-and-switch §13/AC1: a second start, with carried.postgres present, never opens the database and
+// writes nothing.
+func TestASecondStartCarriesNothing(t *testing.T) {
+	mm := newMesh(t)
+	carried := newMemState(func() time.Time { return mm.now })
+	if _, err := carryFor(mm, carried, todaysRows(t, mm)).Run(context.Background(), mm.m, mm.store); err != nil {
+		t.Fatal(err)
+	}
+	writes := carried.writes + mm.licences.writes + mm.bindings.writes + mm.offered.writes
+	entries := len(mm.audit.entries) + len(mm.usage.entries)
+	puts := len(mm.grants.puts)
+	opened := false
+	again := &Carry{State: carried, Source: func(context.Context) (CarrySource, error) { opened = true; return nil, errors.New("no") }}
+	if rec, err := again.Run(context.Background(), mm.m, mm.store); rec != nil || err != nil || opened {
+		t.Fatalf("a second start answered %v %v, opened the database: %v", rec, err, opened)
+	}
+	if carried.writes+mm.licences.writes+mm.bindings.writes+mm.offered.writes != writes ||
+		len(mm.audit.entries)+len(mm.usage.entries) != entries || len(mm.grants.puts) != puts {
+		t.Fatal("a second start wrote")
+	}
+}
+
+// design 39 bus-and-switch §13/AC2: a carry-over stopped after bindings leaves no key, names its table, and a start
+// again carries over whole: every count equals the rows read, and no log entry is appended twice.
+func TestAStoppedCarryOverIsDoneAgainWhole(t *testing.T) {
+	mm := newMesh(t)
+	carried := newMemState(func() time.Time { return mm.now })
+	src := todaysRows(t, mm)
+	src.failAt = "offered"
+	c := carryFor(mm, carried, src)
+	if _, err := c.Run(context.Background(), mm.m, mm.store); err == nil || !strings.Contains(c.Failure(), "offered") {
+		t.Fatalf("a stopped carry-over answered %v, its failure %q", err, c.Failure())
+	}
+	if e, _ := carried.Get(CarriedKey); e != nil || c.Ended() {
+		t.Fatal("a stopped carry-over wrote its key")
+	}
+	src.failAt = ""
+	since := mm.bound("laptop").Since
+	mm.now = mm.now.Add(time.Minute)
+	if _, err := c.Run(context.Background(), mm.m, mm.store); err != nil {
+		t.Fatal(err)
+	}
+	for table, got := range readCarried(t, carried).Tables {
+		if table != "lease" && got.Written != got.Read {
+			t.Errorf("%s: read %d written %d", table, got.Read, got.Written)
+		}
+	}
+	if b := mm.bound("laptop"); b.Since != since {
+		t.Fatalf("the binding put again changed its since from %s to %s", since, b.Since)
+	}
+	if n := len(mm.audit.under("")); n != 5 {
+		t.Fatalf("the audit holds %d entries, want the four rows and one carried", n)
+	}
+	// the run stopped before the logs; a second pass over a log already written appends nothing.
+	co := carryOver{m: mm.m, s: mm.store, src: src, now: mm.now, tables: map[string]CarriedTable{}, audit: src.audit}
+	if _, err := co.usage(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if got, err := co.audits(); err != nil || got.Written != 4 || len(mm.audit.under("")) != 5 || len(mm.usage.under("")) != 2 {
+		t.Fatalf("a log written again appended: %+v %v, %d usage entries", got, err, len(mm.usage.under("")))
+	}
+}
+
+// design 39 bus-and-switch §11/AC2: with the database unreachable the carry-over fails naming it, every acting verb
+// answers "the carry-over has not ended", a read verb is served, and no vendor call is made.
+func TestAnUnreachableDatabaseActsNotAndCallsNoVendor(t *testing.T) {
+	mm := newMesh(t)
+	carried := newMemState(func() time.Time { return mm.now })
+	mm.m.Carry = &Carry{State: carried, Source: func(context.Context) (CarrySource, error) {
+		return nil, errors.New("dial tcp 192.0.2.1:5432: connect: connection refused")
+	}}
+	if _, err := mm.m.Carry.Run(context.Background(), mm.m, mm.store); err == nil || !strings.Contains(mm.m.Carry.Failure(), "connection refused") {
+		t.Fatalf("the run answered %v, its failure %q", err, mm.m.Carry.Failure())
+	}
+	built = mm.m
+	t.Cleanup(func() { built = nil })
+	served := map[string]stdTool{}
+	for _, tool := range tools() {
+		served[strings.TrimPrefix(tool.Name, Seat+".")] = stdTool(tool.Run)
+	}
+	for _, name := range []string{"bind", "switch", "release", "refresh", "adopt"} {
+		if _, err := served[name](map[string]any{"consumer": "laptop", "licence": licence1, "name": "api", "file": "/none"}); err == nil || err.Error() != NotCarried {
+			t.Errorf("%s answered %v", name, err)
+		}
+	}
+	for _, name := range []string{"licences", "bindings", "usage", "public-key"} {
+		if _, err := served[name](map[string]any{}); err != nil {
+			t.Errorf("the read verb %s answered %v", name, err)
+		}
+	}
+	if len(mm.vendor.exchanged) != 0 {
+		t.Fatalf("the vendor was called: %v", mm.vendor.exchanged)
+	}
+}
+
+type stdTool func(map[string]any) (any, error)
+
+// design 39 bus-and-switch §11/AC3: a table that does not exist (a fresh install) is read as no rows; any other
+// failure of the database is one.
+func TestAMissingTableIsNoRows(t *testing.T) {
+	if !noTable(&pgconn.PgError{Code: "42P01", Message: `relation "offered" does not exist`}) {
+		t.Fatal("a missing table is a failure")
+	}
+	for _, err := range []error{&pgconn.PgError{Code: "28P01"}, errors.New("connection refused"), nil} {
+		if noTable(err) {
+			t.Fatalf("%v is read as a missing table", err)
+		}
+	}
+}
+
+// design 39 bus-and-switch §1/AC3: no code of the manager writes to its database.
+func TestNoWriteToTheDatabase(t *testing.T) {
+	write := regexp.MustCompile(`(?i)\b(insert\s+into|update\s+\w+\s+set|delete\s+from|create\s+(table|sequence|index)|drop\s+(table|sequence|index)|alter\s+table|truncate\s|setval\s*\(|nextval\s*\()`)
+	files, _ := os.ReadDir(".")
+	for _, f := range files {
+		if !strings.HasSuffix(f.Name(), ".go") || strings.HasSuffix(f.Name(), "_test.go") {
+			continue
+		}
+		raw, err := os.ReadFile(f.Name())
+		if err != nil {
+			t.Fatal(err)
+		}
+		if w := write.Find(raw); w != nil {
+			t.Errorf("%s writes to the database: %q", f.Name(), w)
+		}
+	}
+}
@@ -73,6 +73,10 @@ func manager() (*Manager, error) {
 			AuditLog: stdio.Log("audit"), Now: time.Now},
 		Vendor: LiveVendor(),
 		Crypt:  crypt,
+		// Today's database is the carry-over's source alone, read once and never written (§11, A3).
+		Carry: &Carry{State: stdio.State("carried"), Source: func(ctx context.Context) (CarrySource, error) {
+			return PgSourceFromEnv(ctx)
+		}},
 		Grants: VaultGrants{Keys: keys},
 		Keys:   keys,
 		AskGrant: func(_ context.Context, node, publicKey string) (GrantAnswer, error) {
@@ -130,7 +134,25 @@ var (
 	heard   = map[string]Holdings{}
 )

+// carryOverFirst runs the carry-over at the start, before the manager hears an event, refreshes or reads usage (§11), and
+// answers once it has ended: at once when `carried.postgres` exists, else when a run writes it, this one or one asked
+// by hand later.
+func carryOverFirst() {
+	<-stdio.Served()
+	m, err := manager()
+	if err != nil {
+		say("the carry-over cannot start: %v", err)
+		return
+	}
+	if !m.Carry.Ended() {
+		bus, _ := m.Store.(*BusStore)
+		_, _ = m.Carry.Run(context.Background(), m, bus) // a failure is said, and the run is asked again by hand (§44)
+	}
+	<-m.Carry.Wait()
+}
+
 func daemon() {
+	carryOverFirst()
 	mu, reports := &heardMu, heard
 	var passing sync.Mutex
 	pass := func() {
@@ -234,7 +256,8 @@ func text(a map[string]any, k string) (string, error) {
 }

 // verb is one of the seat's verbs: listed as `<seat>.<verb>`, so the runtime serves it on the seat's subject.
-// What it changes is recorded as asked by its caller, as the runtime proved it (novox/hq issues 365 and 625).
+// What it changes is recorded as asked by its caller, as the runtime proved it (novox/hq issues 365 and 625). A verb
+// that acts answers nothing but "the carry-over has not ended" until it has (§11).
 func verb(name, description string, input map[string]any, run func(ctx context.Context, m *Manager, a map[string]any) (any, error)) stdio.Tool {
 	return stdio.Tool{Name: Seat + "." + name, Description: description, Input: input,
 		Run: func(a map[string]any) (any, error) {
@@ -242,11 +265,19 @@ func verb(name, description string, input map[string]any, run func(ctx context.C
 			if err != nil {
 				return nil, err
 			}
+			if !reads[name] {
+				if err := m.Carry.Acts(); err != nil {
+					return nil, err
+				}
+			}
 			ctx := WithCause(context.Background(), Cause{Why: "the " + name + " verb", By: stdio.CallerOf(a)})
 			return run(ctx, m, a)
 		}}
 }

+// reads are the seat's verbs that only read, served while the carry-over has not ended (§11).
+var reads = map[string]bool{"licences": true, "bindings": true, "usage": true, "public-key": true, "current": true}
+
 func str(description string) map[string]any {
 	return map[string]any{"type": "string", "description": description}
 }
@@ -155,6 +155,8 @@ type Manager struct {
 	Vendor Vendor
 	// Crypt opens the grants today's database keeps at rest, for the carry-over alone (design 39 bus-and-switch §12, A3).
 	Crypt *Crypt
+	// Carry is the carry-over's gate (§11): no act until it has ended. Nil is a manager with nothing to carry over.
+	Carry *Carry
 	// Grants is the vault, which holds every licence's grant and access token (§15, §16).
 	Grants Grants
 	Keys   KeyPair
@@ -457,10 +459,7 @@ func (m *Manager) adoptOne(ctx context.Context, c Candidate, reports []Holdings,
 // bus-and-switch §15), and keeps on the licence its fingerprint and expiries alone: no token once it returns.
 func (m *Manager) keep(ctx context.Context, l *Licence, g FullGrant) error {
 	grant, _ := json.Marshal(g)
-	access, _ := json.Marshal(map[string]any{"accessToken": g.AccessToken, "expiresAt": g.ExpiresAt,
-		"refreshTokenExpiresAt": g.RefreshTokenExpiresAt, "scopes": g.Scopes,
-		"subscriptionType": g.SubscriptionType, "rateLimitTier": g.RateLimitTier})
-	if err := m.handOver(ctx, l, string(grant), string(access)); err != nil {
+	if err := m.handOver(ctx, l, string(grant), accessOf(g)); err != nil {
 		return err
 	}
 	l.RefreshFingerprint = Fingerprint(g.RefreshToken)
@@ -473,6 +472,14 @@ func (m *Manager) keep(ctx context.Context, l *Licence, g FullGrant) error {
 	return nil
 }

+// accessOf is a subscription's `access` secret: its access token and expiry, never its refresh token (§15).
+func accessOf(g FullGrant) string {
+	access, _ := json.Marshal(map[string]any{"accessToken": g.AccessToken, "expiresAt": g.ExpiresAt,
+		"refreshTokenExpiresAt": g.RefreshTokenExpiresAt, "scopes": g.Scopes,
+		"subscriptionType": g.SubscriptionType, "rateLimitTier": g.RateLimitTier})
+	return string(access)
+}
+
 // handOver puts a licence's two secrets in the vault at its next generation, the grant first (§15).
 func (m *Manager) handOver(ctx context.Context, l *Licence, grant, access string) error {
 	generation := l.Generation + 1
@@ -63,7 +63,9 @@ func (g *memGrants) Put(_ context.Context, key, part string, generation int64, v
 	g.mu.Lock()
 	defer g.mu.Unlock()
 	name := SecretName(key, part)
-	if h, ok := g.held[name]; ok && h.generation >= generation {
+	if h, ok := g.held[name]; ok && h.generation == generation && h.value == value {
+		return nil // the vault keeps what it holds, unchanged
+	} else if ok && h.generation >= generation {
 		return fmt.Errorf("%s is held at generation %d, and generation %d is not newer", name, h.generation, generation)
 	}
 	g.held[name] = heldGrant{generation, value}
@@ -1,71 +1,27 @@
 package main

-// The store on the mesh's postgres (novox/hq design 39 §1), the database the mesh provisioned for this
-// module. The schema is brought to this version's shape by the preparation step (ADR 0135), each
-// statement idempotent.
+// Today's database, read once by the carry-over and never written (design 39 bus-and-switch §1, §11, A3): the
+// module's own, opened with the credential the mesh still provisions for it alone. Every query is a select; a table
+// that does not exist (a fresh install) is read as no rows.

 import (
 	"context"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"os"
 	"strings"
 	"time"

 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgxpool"
 )

-// Schema is what this version needs.
-var Schema = []string{
-	`create table if not exists licence (
-	   name text primary key,
-	   kind text not null check (kind in ('subscription', 'api-key')),
-	   account_uuid text unique,
-	   email text,
-	   organization_uuid text,
-	   sealed text,
-	   refresh_fingerprint text,
-	   access_expires_at bigint,
-	   refresh_expires_at bigint,
-	   failures integer not null default 0,
-	   notified_at bigint,
-	   adopted_at bigint not null,
-	   rotated_at bigint)`,
-	`create sequence if not exists binding_generation`,
-	`create table if not exists binding (
-	   consumer text primary key,
-	   licence text not null references licence(name),
-	   generation bigint not null)`,
-	`create table if not exists lease (
-	   key text primary key,
-	   holder text not null,
-	   until timestamptz not null)`,
-	`create table if not exists offered (
-	   fingerprint text primary key,
-	   node text not null,
-	   account_uuid text,
-	   outcome text not null,
-	   why text not null,
-	   at timestamptz not null default now())`,
-	`create table if not exists usage (
-	   licence text not null,
-	   at bigint not null,
-	   reading jsonb not null,
-	   raw jsonb not null)`,
-	`create index if not exists usage_by_licence on usage (licence, at desc)`,
-	`create table if not exists audit (
-	   at timestamptz not null default now(),
-	   what text not null,
-	   detail jsonb not null)`,
-}
+// PgSource reads today's database.
+type PgSource struct{ pool *pgxpool.Pool }

-// PgStore is the store on postgres.
-type PgStore struct{ pool *pgxpool.Pool }
-
-// PgStoreFromEnv opens the store the mesh provisioned, its URL in the file DATABASE_URL_FILE names.
-func PgStoreFromEnv(ctx context.Context) (*PgStore, error) {
+// PgSourceFromEnv opens the database the mesh provisioned, its URL in the file DATABASE_URL_FILE names.
+func PgSourceFromEnv(ctx context.Context) (*PgSource, error) {
 	file := os.Getenv("DATABASE_URL_FILE")
 	if file == "" {
 		return nil, errors.New("DATABASE_URL_FILE is not set: the manager's database is a requirement the mesh resolves")
@@ -74,219 +30,131 @@ func PgStoreFromEnv(ctx context.Context) (*PgStore, error) {
 	if err != nil {
 		return nil, err
 	}
-	return OpenPgStore(ctx, strings.TrimSpace(string(raw)))
+	return OpenPgSource(ctx, strings.TrimSpace(string(raw)))
 }

-// OpenPgStore opens a store at a URL.
-func OpenPgStore(ctx context.Context, url string) (*PgStore, error) {
+// OpenPgSource opens a database at a URL, and reaches it once, so an unreachable one fails here.
+func OpenPgSource(ctx context.Context, url string) (*PgSource, error) {
 	pool, err := pgxpool.New(ctx, url)
 	if err != nil {
 		return nil, err
 	}
-	return &PgStore{pool: pool}, nil
+	if err := pool.Ping(ctx); err != nil {
+		pool.Close()
+		return nil, err
+	}
+	return &PgSource{pool: pool}, nil
 }

-// Migrate brings the schema to this version's shape.
-func (s *PgStore) Migrate(ctx context.Context) error {
-	for _, q := range Schema {
-		if _, err := s.pool.Exec(ctx, q); err != nil {
-			return fmt.Errorf("%s: %w", strings.Fields(q)[0:6], err)
+// Close lets the database go.
+func (s *PgSource) Close() { s.pool.Close() }
+
+// noTable is whether err is postgres saying a table does not exist (SQLSTATE 42P01).
+func noTable(err error) bool {
+	var pg *pgconn.PgError
+	return errors.As(err, &pg) && pg.Code == "42P01"
+}
+
+// each runs a select and scans each row; a table that does not exist is no rows.
+func (s *PgSource) each(ctx context.Context, q string, scan func(pgx.Rows) error, args ...any) error {
+	rows, err := s.pool.Query(ctx, q, args...)
+	if err != nil {
+		if noTable(err) {
+			return nil
 		}
+		return err
+	}
+	defer rows.Close()
+	for rows.Next() {
+		if err := scan(rows); err != nil {
+			return err
+		}
+	}
+	if err := rows.Err(); err != nil && !noTable(err) {
+		return err
 	}
 	return nil
 }

-const licenceColumns = `name, kind, coalesce(account_uuid,''), coalesce(email,''), coalesce(organization_uuid,''),
-  coalesce(sealed,''), coalesce(refresh_fingerprint,''), coalesce(access_expires_at,0), coalesce(refresh_expires_at,0),
-  failures, coalesce(notified_at,0), adopted_at, coalesce(rotated_at,0)`
-
-func scanLicence(row pgx.Row) (*Licence, error) {
-	var l Licence
-	var sealed string // the grant at rest, read by the carry-over alone (design 39 bus-and-switch §12)
-	err := row.Scan(&l.Name, &l.Kind, &l.AccountUUID, &l.Email, &l.OrganizationUUID, &sealed, &l.RefreshFingerprint,
-		&l.AccessExpiresAt, &l.RefreshExpiresAt, &l.Failures, &l.NotifiedAt, &l.AdoptedAt, &l.RotatedAt)
-	if errors.Is(err, pgx.ErrNoRows) {
-		return nil, nil
-	}
-	return &l, err
+func (s *PgSource) Licences(ctx context.Context) ([]LicenceRow, error) {
+	out := []LicenceRow{}
+	err := s.each(ctx, `select name, kind, coalesce(account_uuid,''), coalesce(email,''), coalesce(organization_uuid,''),
+	    coalesce(sealed,''), coalesce(refresh_fingerprint,''), coalesce(access_expires_at,0),
+	    coalesce(refresh_expires_at,0), failures, coalesce(notified_at,0), adopted_at, coalesce(rotated_at,0)
+	  from licence order by adopted_at, name`, func(r pgx.Rows) error {
+		var l LicenceRow
+		err := r.Scan(&l.Name, &l.Kind, &l.AccountUUID, &l.Email, &l.OrganizationUUID, &l.Sealed, &l.RefreshFingerprint,
+			&l.AccessExpiresAt, &l.RefreshExpiresAt, &l.Failures, &l.NotifiedAt, &l.AdoptedAt, &l.RotatedAt)
+		out = append(out, l)
+		return err
+	})
+	return out, err
 }

-func (s *PgStore) Licences(ctx context.Context) ([]Licence, error) {
-	rows, err := s.pool.Query(ctx, `select `+licenceColumns+` from licence order by name`)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-	var out []Licence
-	for rows.Next() {
-		l, err := scanLicence(rows)
-		if err != nil {
-			return nil, err
+func (s *PgSource) Bindings(ctx context.Context) ([]Binding, error) {
+	out := []Binding{}
+	err := s.each(ctx, `select consumer, licence, generation from binding order by generation, consumer`,
+		func(r pgx.Rows) error {
+			var b Binding
+			err := r.Scan(&b.Consumer, &b.Licence, &b.Generation)
+			out = append(out, b)
+			return err
+		})
+	return out, err
+}
+
+func (s *PgSource) Leases(ctx context.Context) (int, error) {
+	n := 0
+	err := s.each(ctx, `select key from lease`, func(r pgx.Rows) error {
+		n++
+		var k string
+		return r.Scan(&k)
+	})
+	return n, err
+}
+
+func (s *PgSource) Offered(ctx context.Context) ([]OfferedRow, error) {
+	out := []OfferedRow{}
+	err := s.each(ctx, `select fingerprint, node, coalesce(account_uuid,''), outcome, why, at from offered
+	  order by at, fingerprint`, func(r pgx.Rows) error {
+		var o OfferedRow
+		var outcome string
+		err := r.Scan(&o.Fingerprint, &o.Node, &o.Account, &outcome, &o.Why, &o.At)
+		o.Outcome = Outcome(outcome)
+		out = append(out, o)
+		return err
+	})
+	return out, err
+}
+
+func (s *PgSource) Usage(ctx context.Context, since time.Time) ([]UsageTableRow, error) {
+	out := []UsageTableRow{}
+	err := s.each(ctx, `select licence, at, reading, raw from usage where at >= $1 order by at, licence`,
+		func(r pgx.Rows) error {
+			var u UsageTableRow
+			var reading, raw []byte
+			if err := r.Scan(&u.Licence, &u.At, &reading, &raw); err != nil {
+				return err
+			}
+			_ = json.Unmarshal(reading, &u.Reading)
+			_ = json.Unmarshal(raw, &u.Raw)
+			out = append(out, u)
+			return nil
+		}, since.UnixMilli())
+	return out, err
+}
+
+func (s *PgSource) Audit(ctx context.Context) ([]AuditRow, error) {
+	out := []AuditRow{}
+	err := s.each(ctx, `select at, what, detail from audit order by at`, func(r pgx.Rows) error {
+		var a AuditRow
+		var detail []byte
+		if err := r.Scan(&a.At, &a.What, &detail); err != nil {
+			return err
 		}
-		out = append(out, *l)
-	}
-	return out, rows.Err()
-}
-
-func (s *PgStore) Licence(ctx context.Context, name string) (*Licence, error) {
-	return scanLicence(s.pool.QueryRow(ctx, `select `+licenceColumns+` from licence where name = $1`, name))
-}
-
-func (s *PgStore) LicenceForAccount(ctx context.Context, account string) (*Licence, error) {
-	return scanLicence(s.pool.QueryRow(ctx, `select `+licenceColumns+` from licence where account_uuid = $1`, account))
-}
-
-func nullable(s string) any {
-	if s == "" {
+		_ = json.Unmarshal(detail, &a.Detail)
+		out = append(out, a)
 		return nil
-	}
-	return s
+	})
+	return out, err
 }
-
-func nullableInt(v int64) any {
-	if v == 0 {
-		return nil
-	}
-	return v
-}
-
-func (s *PgStore) SaveLicence(ctx context.Context, l Licence) error {
-	_, err := s.pool.Exec(ctx, `insert into licence (name, kind, account_uuid, email, organization_uuid, refresh_fingerprint,
-	    access_expires_at, refresh_expires_at, failures, notified_at, adopted_at, rotated_at)
-	  values ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)
-	  on conflict (name) do update set kind = excluded.kind, account_uuid = excluded.account_uuid, email = excluded.email,
-	    organization_uuid = excluded.organization_uuid, refresh_fingerprint = excluded.refresh_fingerprint,
-	    access_expires_at = excluded.access_expires_at, refresh_expires_at = excluded.refresh_expires_at,
-	    failures = excluded.failures, notified_at = excluded.notified_at, adopted_at = excluded.adopted_at,
-	    rotated_at = excluded.rotated_at`,
-		l.Name, l.Kind, nullable(l.AccountUUID), nullable(l.Email), nullable(l.OrganizationUUID),
-		nullable(l.RefreshFingerprint), nullableInt(l.AccessExpiresAt), nullableInt(l.RefreshExpiresAt), l.Failures,
-		nullableInt(l.NotifiedAt), l.AdoptedAt, nullableInt(l.RotatedAt))
-	return err
-}
-
-// Lease is taken in the store before a row is read (design 39 §3): a second run started together finds
-// it live and does nothing.
-func (s *PgStore) Lease(ctx context.Context, key, holder string, d time.Duration) (bool, error) {
-	tag, err := s.pool.Exec(ctx, `insert into lease (key, holder, until) values ($1, $2, now() + make_interval(secs => $3))
-	  on conflict (key) do update set holder = excluded.holder, until = excluded.until
-	    where lease.until < now() or lease.holder = excluded.holder`, key, holder, d.Seconds())
-	if err != nil {
-		return false, err
-	}
-	return tag.RowsAffected() == 1, nil
-}
-
-func (s *PgStore) Unlease(ctx context.Context, key, holder string) error {
-	_, err := s.pool.Exec(ctx, `delete from lease where key = $1 and holder = $2`, key, holder)
-	return err
-}
-
-func (s *PgStore) bindingsWhere(ctx context.Context, q string, args ...any) ([]Binding, error) {
-	rows, err := s.pool.Query(ctx, q, args...)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-	var out []Binding
-	for rows.Next() {
-		var b Binding
-		if err := rows.Scan(&b.Consumer, &b.Licence, &b.Generation); err != nil {
-			return nil, err
-		}
-		out = append(out, b)
-	}
-	return out, rows.Err()
-}
-
-func (s *PgStore) Bindings(ctx context.Context) ([]Binding, error) {
-	return s.bindingsWhere(ctx, `select consumer, licence, generation from binding order by consumer`)
-}
-
-func (s *PgStore) Binding(ctx context.Context, consumer string) (*Binding, error) {
-	out, err := s.bindingsWhere(ctx, `select consumer, licence, generation from binding where consumer = $1`, consumer)
-	if err != nil || len(out) == 0 {
-		return nil, err
-	}
-	return &out[0], nil
-}
-
-func (s *PgStore) Bind(ctx context.Context, consumer, licence string) (Binding, error) {
-	out, err := s.bindingsWhere(ctx, `insert into binding (consumer, licence, generation) values ($1, $2, nextval('binding_generation'))
-	  on conflict (consumer) do update set licence = excluded.licence, generation = excluded.generation
-	  returning consumer, licence, generation`, consumer, licence)
-	if err != nil {
-		return Binding{}, err
-	}
-	return out[0], nil
-}
-
-func (s *PgStore) Unbind(ctx context.Context, consumer string) (bool, error) {
-	tag, err := s.pool.Exec(ctx, `delete from binding where consumer = $1`, consumer)
-	return err == nil && tag.RowsAffected() == 1, err
-}
-
-func (s *PgStore) GenerationsAbove(ctx context.Context, generation int64) error {
-	if generation <= 0 {
-		return nil
-	}
-	// setval with is_called, so the next nextval is generation+1; only ever forward.
-	_, err := s.pool.Exec(ctx, `select setval('binding_generation', $1, true) from binding_generation
-	  where not is_called or last_value < $1`, generation)
-	return err
-}
-
-func (s *PgStore) Advance(ctx context.Context, licence string) ([]Binding, error) {
-	return s.bindingsWhere(ctx, `update binding set generation = nextval('binding_generation') where licence = $1
-	  returning consumer, licence, generation`, licence)
-}
-
-func (s *PgStore) Outcome(ctx context.Context, fp string) (Outcome, error) {
-	var o string
-	err := s.pool.QueryRow(ctx, `select outcome from offered where fingerprint = $1`, fp).Scan(&o)
-	if errors.Is(err, pgx.ErrNoRows) {
-		return "", nil
-	}
-	return Outcome(o), err
-}
-
-func (s *PgStore) RecordOutcome(ctx context.Context, fp, node, account string, o Outcome, why string) error {
-	_, err := s.pool.Exec(ctx, `insert into offered (fingerprint, node, account_uuid, outcome, why) values ($1,$2,$3,$4,$5)
-	  on conflict (fingerprint) do update set outcome = excluded.outcome, why = excluded.why, at = now()`,
-		fp, node, nullable(account), string(o), why)
-	return err
-}
-
-func (s *PgStore) RecordUsage(ctx context.Context, licence string, at int64, r UsageReading, raw map[string]any) error {
-	reading, _ := json.Marshal(r)
-	rawJSON, _ := json.Marshal(raw)
-	_, err := s.pool.Exec(ctx, `insert into usage (licence, at, reading, raw) values ($1,$2,$3,$4)`, licence, at, reading, rawJSON)
-	return err
-}
-
-func (s *PgStore) Usage(ctx context.Context, licence string, limit int) ([]UsageRow, error) {
-	rows, err := s.pool.Query(ctx, `select licence, at, reading from usage where ($1 = '' or licence = $1) order by at desc limit $2`, licence, limit)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-	var out []UsageRow
-	for rows.Next() {
-		var u UsageRow
-		var reading []byte
-		if err := rows.Scan(&u.Licence, &u.At, &reading); err != nil {
-			return nil, err
-		}
-		_ = json.Unmarshal(reading, &u.Reading)
-		out = append(out, u)
-	}
-	return out, rows.Err()
-}
-
-func (s *PgStore) Audit(ctx context.Context, what string, detail map[string]any) error {
-	raw, _ := json.Marshal(detail)
-	_, err := s.pool.Exec(ctx, `insert into audit (what, detail) values ($1, $2)`, what, raw)
-	return err
-}
-
-func (s *PgStore) Close() { s.pool.Close() }
@@ -1,97 +1,12 @@
 package main

 import (
-	"context"
 	"crypto/rand"
 	"encoding/base64"
-	"os"
 	"strings"
 	"testing"
-	"time"
 )

-// Against a real postgres, because the questions are the database's: does the schema apply twice, does a
-// lease refuse a second holder, does a generation only grow. Skipped unless one is named:
-//
-//	docker run -d --rm --name licmgr-pg -e POSTGRES_PASSWORD=t -p 15498:5432 postgres:16-alpine
-//	MESH_TEST_POSTGRES=postgres://postgres:t@127.0.0.1:15498/postgres go test ./...
-func TestTheStoreOnPostgres(t *testing.T) {
-	url := os.Getenv("MESH_TEST_POSTGRES")
-	if url == "" {
-		t.Skip("MESH_TEST_POSTGRES unset")
-	}
-	ctx := context.Background()
-	s, err := OpenPgStore(ctx, url)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer s.Close()
-	for _, table := range []string{"binding", "licence", "lease", "offered", "usage", "audit"} {
-		_, _ = s.pool.Exec(ctx, "drop table if exists "+table+" cascade")
-	}
-	_, _ = s.pool.Exec(ctx, "drop sequence if exists binding_generation")
-	for i := 0; i < 2; i++ {
-		if err := s.Migrate(ctx); err != nil {
-			t.Fatalf("migration %d: %v", i+1, err)
-		}
-	}
-	l := Licence{Name: "a@example.org", Kind: "subscription", AccountUUID: "u-1", Email: "a@example.org",
-		RefreshFingerprint: "sha256:1", AccessExpiresAt: 1, RefreshExpiresAt: 2, AdoptedAt: 3}
-	if err := s.SaveLicence(ctx, l); err != nil {
-		t.Fatal(err)
-	}
-	l.Failures = 2
-	_ = s.SaveLicence(ctx, l)
-	if got, _ := s.LicenceForAccount(ctx, "u-1"); got == nil || got.Failures != 2 || got.OrganizationUUID != "" {
-		t.Fatalf("%+v", got)
-	}
-	if none, err := s.Licence(ctx, "nobody"); none != nil || err != nil {
-		t.Fatalf("%v %v", none, err)
-	}
-	lease := func(holder string) bool {
-		ok, err := s.Lease(ctx, "licence:a", holder, time.Minute)
-		if err != nil {
-			t.Fatal(err)
-		}
-		return ok
-	}
-	if !lease("one") || lease("two") || !lease("one") {
-		t.Fatal("a lease did not refuse a second holder, or its own holder could not renew it")
-	}
-	_ = s.Unlease(ctx, "licence:a", "one")
-	if !lease("two") {
-		t.Fatal("a released lease was not taken")
-	}
-	b1, _ := s.Bind(ctx, "laptop", l.Name)
-	adv, _ := s.Advance(ctx, l.Name)
-	b3, _ := s.Bind(ctx, "laptop", l.Name)
-	if len(adv) != 1 || !(b1.Generation < adv[0].Generation && adv[0].Generation < b3.Generation) {
-		t.Fatalf("generations %d %v %d", b1.Generation, adv, b3.Generation)
-	}
-	_ = s.RecordOutcome(ctx, "sha256:x", "laptop", "u-1", Dead, "400")
-	if o, _ := s.Outcome(ctx, "sha256:x"); o != Dead {
-		t.Fatalf("outcome %q", o)
-	}
-	if o, _ := s.Outcome(ctx, "sha256:none"); o != "" {
-		t.Fatalf("an unknown login is %q", o)
-	}
-	pct := 1.0
-	_ = s.RecordUsage(ctx, l.Name, 5, UsageReading{SessionPct: &pct}, map[string]any{})
-	if u, _ := s.Usage(ctx, "", 5); len(u) != 1 || *u[0].Reading.SessionPct != 1 {
-		t.Fatalf("usage %v", u)
-	}
-	if err := s.Audit(ctx, "bound", map[string]any{"consumer": "laptop"}); err != nil {
-		t.Fatal(err)
-	}
-	if ok, _ := s.Unbind(ctx, "laptop"); !ok {
-		t.Fatal("unbind")
-	}
-	all, _ := s.Licences(ctx)
-	if len(all) != 1 {
-		t.Fatalf("%v", all)
-	}
-}
-
 func TestARefreshThatDoesNotRotateKeepsTheRefreshToken(t *testing.T) {
 	prev := FullGrant{AccessToken: "a", RefreshToken: "r", ExpiresAt: 1}
 	in := int64(60)