M-6 fix: the carry-over carries each live lease with its remaining lifetime (issue 1409)

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

modules/claude-licence-manager/cmd/claude-licence-manager/carry.go

@@ -10,6 +10,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
 	"sync"
 	"time"
 )
@@ -59,12 +60,19 @@ type AuditRow struct {
 type CarrySource interface {
 	Licences(ctx context.Context) ([]LicenceRow, 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)
 	Usage(ctx context.Context, since time.Time) ([]UsageTableRow, 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.
 type CarriedTable struct {
 	Read    int `json:"read"`
@@ -219,7 +227,7 @@ func (co *carryOver) all(ctx context.Context) error {
 		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
+			t, err = co.leases(ctx)
 		case "offered":
 			t, err = co.offered(ctx)
 		case "usage":
@@ -334,6 +342,38 @@ func (co *carryOver) bindings(ctx context.Context) (CarriedTable, error) {
 	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
 // gone already and written nowhere (§5, §12).
 func (co *carryOver) offered(ctx context.Context) (CarriedTable, error) {

modules/claude-licence-manager/cmd/claude-licence-manager/carry_test.go

@@ -18,7 +18,7 @@ import (
 type memSource struct {
 	licences []LicenceRow
 	bindings []Binding
-	leases   int
+	leases   []LeaseRow
 	offered  []OfferedRow
 	usage    []UsageTableRow
 	audit    []AuditRow
@@ -37,7 +37,7 @@ func (s *memSource) Licences(context.Context) ([]LicenceRow, error) {
 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) Leases(context.Context) ([]LeaseRow, error) { return s.leases, s.fail("lease") }
 func (s *memSource) Offered(context.Context) ([]OfferedRow, error) {
 	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")},
 		},
 		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)},
 			{Fingerprint: "sha256:bb", Node: "server", Outcome: Dead, Why: "400", At: t0.Add(-48 * time.Hour)}},
 		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)
+	}
+}

modules/claude-licence-manager/cmd/claude-licence-manager/pgstore.go

@@ -103,14 +103,15 @@ func (s *PgSource) Bindings(ctx context.Context) ([]Binding, error) {
 	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)
+func (s *PgSource) Leases(ctx context.Context) ([]LeaseRow, error) {
+	out := []LeaseRow{}
+	err := s.each(ctx, `select key, holder, until from lease order by key`, func(r pgx.Rows) error {
+		var l LeaseRow
+		err := r.Scan(&l.Key, &l.Holder, &l.Until)
+		out = append(out, l)
+		return err
 	})
-	return n, err
+	return out, err
 }
 
 func (s *PgSource) Offered(ctx context.Context) ([]OfferedRow, error) {