M-6 fix: the carry-over carries each live lease with its remaining lifetime (issue 1409)
mesh/delivery merged: source.main moved to novox/mesh-catalog@d9f2b7bcfa07
mesh/delivery-group group feat/39-licence-switch merged: every member is merged
mesh/check checking dbcba161d57c on d9f2b7bcfa07

The lead's answer B44: a lease the bundle before held in today's database
and still live at the run is written to `leases` under the bus's key form
with what is left of its lifetime, so no refresh races it whatever the
delivery's order; an expired one is counted read and not written.

Builds: design 39 bus-and-switch §4, §12, B44

Claude-Session: https://claude.ai/code/session_01NcbgfHb9d4T9nq5Q6WqdiJ
This commit is contained in:
2026-11-12 01:07:09 +01:00
parent c8e321c8e1
commit dbcba161d5
3 changed files with 80 additions and 12 deletions
@@ -10,6 +10,7 @@ import (
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"strings"
"sync" "sync"
"time" "time"
) )
@@ -59,12 +60,19 @@ type AuditRow struct {
type CarrySource interface { type CarrySource interface {
Licences(ctx context.Context) ([]LicenceRow, error) Licences(ctx context.Context) ([]LicenceRow, error)
Bindings(ctx context.Context) ([]Binding, error) Bindings(ctx context.Context) ([]Binding, error)
Leases(ctx context.Context) (int, error) Leases(ctx context.Context) ([]LeaseRow, error)
Offered(ctx context.Context) ([]OfferedRow, error) Offered(ctx context.Context) ([]OfferedRow, error)
Usage(ctx context.Context, since time.Time) ([]UsageTableRow, error) Usage(ctx context.Context, since time.Time) ([]UsageTableRow, error)
Audit(ctx context.Context) ([]AuditRow, error) Audit(ctx context.Context) ([]AuditRow, error)
} }
// LeaseRow is one row of today's table `lease`: its key (`licence:<name>` or `account:<id>`), holder and until when.
type LeaseRow struct {
Key string
Holder string
Until time.Time
}
// CarriedTable is one table in `carried.postgres`: the rows read and the entries written for them. // CarriedTable is one table in `carried.postgres`: the rows read and the entries written for them.
type CarriedTable struct { type CarriedTable struct {
Read int `json:"read"` Read int `json:"read"`
@@ -219,7 +227,7 @@ func (co *carryOver) all(ctx context.Context) error {
case "binding": case "binding":
t, err = co.bindings(ctx) t, err = co.bindings(ctx)
case "lease": case "lease":
t.Read, err = co.src.Leases(ctx) // a lease lives two minutes: none taken before the start still holds t, err = co.leases(ctx)
case "offered": case "offered":
t, err = co.offered(ctx) t, err = co.offered(ctx)
case "usage": case "usage":
@@ -334,6 +342,38 @@ func (co *carryOver) bindings(ctx context.Context) (CarriedTable, error) {
return t, nil return t, nil
} }
// leases writes a key of `leases` per lease still live at the run, with its remaining lifetime, so the manager never
// refreshes a licence the bundle before it holds a lease on, whatever the delivery's order; an expired one is counted
// read and not written (B44). Its key takes the bus's form: `licence.<licence key>` or `account.<id>` (§4).
func (co *carryOver) leases(ctx context.Context) (CarriedTable, error) {
rows, err := co.src.Leases(ctx)
if err != nil {
return CarriedTable{}, err
}
t := CarriedTable{Read: len(rows)}
for _, r := range rows {
left := r.Until.Sub(co.now)
if left <= 0 {
continue
}
key := ""
if name, ok := strings.CutPrefix(r.Key, "licence:"); ok {
key = "licence." + KeyFor(name)
} else if account, ok := strings.CutPrefix(r.Key, "account:"); ok {
key = "account." + account
} else {
co.m.Log("the lease %s names neither a licence nor an account; it is not carried", r.Key)
continue
}
v := leaseValue{Holder: r.Holder, Until: r.Until.UTC().Format(time.RFC3339)}
if _, err := co.s.LeasesState.PutTTL(key, v, left); 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 // 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). // gone already and written nowhere (§5, §12).
func (co *carryOver) offered(ctx context.Context) (CarriedTable, error) { func (co *carryOver) offered(ctx context.Context) (CarriedTable, error) {
@@ -18,7 +18,7 @@ import (
type memSource struct { type memSource struct {
licences []LicenceRow licences []LicenceRow
bindings []Binding bindings []Binding
leases int leases []LeaseRow
offered []OfferedRow offered []OfferedRow
usage []UsageTableRow usage []UsageTableRow
audit []AuditRow audit []AuditRow
@@ -37,7 +37,7 @@ func (s *memSource) Licences(context.Context) ([]LicenceRow, error) {
func (s *memSource) Bindings(context.Context) ([]Binding, error) { func (s *memSource) Bindings(context.Context) ([]Binding, error) {
return s.bindings, s.fail("binding") return s.bindings, s.fail("binding")
} }
func (s *memSource) Leases(context.Context) (int, error) { return s.leases, s.fail("lease") } func (s *memSource) Leases(context.Context) ([]LeaseRow, error) { return s.leases, s.fail("lease") }
func (s *memSource) Offered(context.Context) ([]OfferedRow, error) { func (s *memSource) Offered(context.Context) ([]OfferedRow, error) {
return s.offered, s.fail("offered") return s.offered, s.fail("offered")
} }
@@ -65,7 +65,7 @@ func todaysRows(t *testing.T, mm *miniMesh) *memSource {
{Licence: Licence{Name: "api", Kind: "api-key", AdoptedAt: 2}, Sealed: sealAtRest(t, mm.m.Crypt, "sk-ant-api03-x")}, {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}}, bindings: []Binding{{Consumer: "laptop", Licence: licence1, Generation: 1013}, {Consumer: "server", Licence: "api", Generation: 1016}},
leases: 1, leases: []LeaseRow{{Key: "licence:x", Holder: "novox:1", Until: t0.Add(-time.Minute)}},
offered: []OfferedRow{{Fingerprint: "sha256:aa", Node: "laptop", Account: "acct-1", Outcome: Adopted, Why: "refreshed", At: t0.Add(-24 * time.Hour)}, 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)}}, {Fingerprint: "sha256:bb", Node: "server", Outcome: Dead, Why: "400", At: t0.Add(-48 * time.Hour)}},
usage: []UsageTableRow{ usage: []UsageTableRow{
@@ -285,3 +285,30 @@ func TestNoWriteToTheDatabase(t *testing.T) {
} }
} }
} }
// design 39 bus-and-switch §12, B44: a lease still live at the run is carried with its remaining lifetime, and a
// refresh waits for it to end; an expired one is counted read and not written.
func TestALiveLeaseIsCarriedAndBlocksARefreshUntilItEnds(t *testing.T) {
mm := newMesh(t)
carried := newMemState(func() time.Time { return mm.now })
src := todaysRows(t, mm)
src.leases = []LeaseRow{{Key: "licence:" + licence1, Holder: "the old bundle", Until: mm.now.Add(90 * time.Second)},
{Key: "account:acct-1", Holder: "the old bundle", Until: mm.now.Add(-time.Second)}}
if _, err := carryFor(mm, carried, src).Run(context.Background(), mm.m, mm.store); err != nil {
t.Fatal(err)
}
if got := readCarried(t, carried).Tables["lease"]; got.Read != 2 || got.Written != 1 {
t.Fatalf("lease: read %d written %d, want 2 read and 1 written", got.Read, got.Written)
}
if e, _ := mm.leases.Get("account.acct-1"); e != nil {
t.Fatalf("an expired lease was carried: %s", e.Value)
}
r, err := mm.m.Rotate(context.Background(), licence1, true)
if err != nil || r["refreshed"] != false || r["reason"] != "another run holds its lease" {
t.Fatalf("a refresh under the carried lease answered %v, %v", r, err)
}
mm.now = mm.now.Add(91 * time.Second)
if r, err := mm.m.Rotate(context.Background(), licence1, true); err != nil || r["reason"] == "another run holds its lease" {
t.Fatalf("once the carried lease ended the refresh answered %v, %v", r, err)
}
}
@@ -103,14 +103,15 @@ func (s *PgSource) Bindings(ctx context.Context) ([]Binding, error) {
return out, err return out, err
} }
func (s *PgSource) Leases(ctx context.Context) (int, error) { func (s *PgSource) Leases(ctx context.Context) ([]LeaseRow, error) {
n := 0 out := []LeaseRow{}
err := s.each(ctx, `select key from lease`, func(r pgx.Rows) error { err := s.each(ctx, `select key, holder, until from lease order by key`, func(r pgx.Rows) error {
n++ var l LeaseRow
var k string err := r.Scan(&l.Key, &l.Holder, &l.Until)
return r.Scan(&k) out = append(out, l)
return err
}) })
return n, err return out, err
} }
func (s *PgSource) Offered(ctx context.Context) ([]OfferedRow, error) { func (s *PgSource) Offered(ctx context.Context) ([]OfferedRow, error) {