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
This commit is contained in:
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}, AuditLog: stdio.Log("audit"), Now: time.Now},
Vendor: LiveVendor(), Vendor: LiveVendor(),
Crypt: crypt, 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}, Grants: VaultGrants{Keys: keys},
Keys: keys, Keys: keys,
AskGrant: func(_ context.Context, node, publicKey string) (GrantAnswer, error) { AskGrant: func(_ context.Context, node, publicKey string) (GrantAnswer, error) {
@@ -130,7 +134,25 @@ var (
heard = map[string]Holdings{} 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() { func daemon() {
carryOverFirst()
mu, reports := &heardMu, heard mu, reports := &heardMu, heard
var passing sync.Mutex var passing sync.Mutex
pass := func() { 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. // 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 { 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, return stdio.Tool{Name: Seat + "." + name, Description: description, Input: input,
Run: func(a map[string]any) (any, error) { 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 { if err != nil {
return nil, err 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)}) ctx := WithCause(context.Background(), Cause{Why: "the " + name + " verb", By: stdio.CallerOf(a)})
return run(ctx, m, 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 { func str(description string) map[string]any {
return map[string]any{"type": "string", "description": description} return map[string]any{"type": "string", "description": description}
} }
@@ -155,6 +155,8 @@ type Manager struct {
Vendor Vendor 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 opens the grants today's database keeps at rest, for the carry-over alone (design 39 bus-and-switch §12, A3).
Crypt *Crypt 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 is the vault, which holds every licence's grant and access token (§15, §16).
Grants Grants Grants Grants
Keys KeyPair 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. // 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 { func (m *Manager) keep(ctx context.Context, l *Licence, g FullGrant) error {
grant, _ := json.Marshal(g) grant, _ := json.Marshal(g)
access, _ := json.Marshal(map[string]any{"accessToken": g.AccessToken, "expiresAt": g.ExpiresAt, if err := m.handOver(ctx, l, string(grant), accessOf(g)); err != nil {
"refreshTokenExpiresAt": g.RefreshTokenExpiresAt, "scopes": g.Scopes,
"subscriptionType": g.SubscriptionType, "rateLimitTier": g.RateLimitTier})
if err := m.handOver(ctx, l, string(grant), string(access)); err != nil {
return err return err
} }
l.RefreshFingerprint = Fingerprint(g.RefreshToken) l.RefreshFingerprint = Fingerprint(g.RefreshToken)
@@ -473,6 +472,14 @@ func (m *Manager) keep(ctx context.Context, l *Licence, g FullGrant) error {
return nil 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). // 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 { func (m *Manager) handOver(ctx context.Context, l *Licence, grant, access string) error {
generation := l.Generation + 1 generation := l.Generation + 1
@@ -63,7 +63,9 @@ func (g *memGrants) Put(_ context.Context, key, part string, generation int64, v
g.mu.Lock() g.mu.Lock()
defer g.mu.Unlock() defer g.mu.Unlock()
name := SecretName(key, part) 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) 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} g.held[name] = heldGrant{generation, value}
@@ -1,71 +1,27 @@
package main package main
// The store on the mesh's postgres (novox/hq design 39 §1), the database the mesh provisioned for this // Today's database, read once by the carry-over and never written (design 39 bus-and-switch §1, §11, A3): the
// module. The schema is brought to this version's shape by the preparation step (ADR 0135), each // module's own, opened with the credential the mesh still provisions for it alone. Every query is a select; a table
// statement idempotent. // that does not exist (a fresh install) is read as no rows.
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"errors" "errors"
"fmt"
"os" "os"
"strings" "strings"
"time" "time"
"github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgconn"
"github.com/jackc/pgx/v5/pgxpool" "github.com/jackc/pgx/v5/pgxpool"
) )
// Schema is what this version needs. // PgSource reads today's database.
var Schema = []string{ type PgSource struct{ pool *pgxpool.Pool }
`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)`,
}
// PgStore is the store on postgres. // PgSourceFromEnv opens the database the mesh provisioned, its URL in the file DATABASE_URL_FILE names.
type PgStore struct{ pool *pgxpool.Pool } func PgSourceFromEnv(ctx context.Context) (*PgSource, error) {
// PgStoreFromEnv opens the store the mesh provisioned, its URL in the file DATABASE_URL_FILE names.
func PgStoreFromEnv(ctx context.Context) (*PgStore, error) {
file := os.Getenv("DATABASE_URL_FILE") file := os.Getenv("DATABASE_URL_FILE")
if file == "" { if file == "" {
return nil, errors.New("DATABASE_URL_FILE is not set: the manager's database is a requirement the mesh resolves") 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 { if err != nil {
return nil, err return nil, err
} }
return OpenPgStore(ctx, strings.TrimSpace(string(raw))) return OpenPgSource(ctx, strings.TrimSpace(string(raw)))
} }
// OpenPgStore opens a store at a URL. // OpenPgSource opens a database at a URL, and reaches it once, so an unreachable one fails here.
func OpenPgStore(ctx context.Context, url string) (*PgStore, error) { func OpenPgSource(ctx context.Context, url string) (*PgSource, error) {
pool, err := pgxpool.New(ctx, url) pool, err := pgxpool.New(ctx, url)
if err != nil { if err != nil {
return nil, err 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. // Close lets the database go.
func (s *PgStore) Migrate(ctx context.Context) error { func (s *PgSource) Close() { s.pool.Close() }
for _, q := range Schema {
if _, err := s.pool.Exec(ctx, q); err != nil { // noTable is whether err is postgres saying a table does not exist (SQLSTATE 42P01).
return fmt.Errorf("%s: %w", strings.Fields(q)[0:6], err) 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 return nil
} }
const licenceColumns = `name, kind, coalesce(account_uuid,''), coalesce(email,''), coalesce(organization_uuid,''), func (s *PgSource) Licences(ctx context.Context) ([]LicenceRow, error) {
coalesce(sealed,''), coalesce(refresh_fingerprint,''), coalesce(access_expires_at,0), coalesce(refresh_expires_at,0), out := []LicenceRow{}
failures, coalesce(notified_at,0), adopted_at, coalesce(rotated_at,0)` 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),
func scanLicence(row pgx.Row) (*Licence, error) { coalesce(refresh_expires_at,0), failures, coalesce(notified_at,0), adopted_at, coalesce(rotated_at,0)
var l Licence from licence order by adopted_at, name`, func(r pgx.Rows) error {
var sealed string // the grant at rest, read by the carry-over alone (design 39 bus-and-switch §12) var l LicenceRow
err := row.Scan(&l.Name, &l.Kind, &l.AccountUUID, &l.Email, &l.OrganizationUUID, &sealed, &l.RefreshFingerprint, 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) &l.AccessExpiresAt, &l.RefreshExpiresAt, &l.Failures, &l.NotifiedAt, &l.AdoptedAt, &l.RotatedAt)
if errors.Is(err, pgx.ErrNoRows) { out = append(out, l)
return nil, nil return err
} })
return &l, err return out, err
} }
func (s *PgStore) Licences(ctx context.Context) ([]Licence, error) { func (s *PgSource) Bindings(ctx context.Context) ([]Binding, error) {
rows, err := s.pool.Query(ctx, `select `+licenceColumns+` from licence order by name`) out := []Binding{}
if err != nil { err := s.each(ctx, `select consumer, licence, generation from binding order by generation, consumer`,
return nil, err func(r pgx.Rows) error {
} var b Binding
defer rows.Close() err := r.Scan(&b.Consumer, &b.Licence, &b.Generation)
var out []Licence out = append(out, b)
for rows.Next() { return err
l, err := scanLicence(rows) })
if err != nil { return out, err
return nil, 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) _ = json.Unmarshal(detail, &a.Detail)
} out = append(out, a)
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 == "" {
return nil 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 package main
import ( import (
"context"
"crypto/rand" "crypto/rand"
"encoding/base64" "encoding/base64"
"os"
"strings" "strings"
"testing" "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) { func TestARefreshThatDoesNotRotateKeepsTheRefreshToken(t *testing.T) {
prev := FullGrant{AccessToken: "a", RefreshToken: "r", ExpiresAt: 1} prev := FullGrant{AccessToken: "a", RefreshToken: "r", ExpiresAt: 1}
in := int64(60) in := int64(60)