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
This commit is contained in:
@@ -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) {
|
||||||
|
|||||||
Reference in New Issue
Block a user