amadan
moves: reservation kinds for creates and placements, and a table that is never deleted, never repeats a body, is normalised before any acquisition and is written one HTTP attempt at a time, resent until answered with each registration removed only by its owner, and is read without a write where a box only looks
Task 2a of self-serve moves plan 4. release gains SingleAttemptPutter, PutOnce and ErrNoSingleAttemptPut (S3Store sends one attempt through a per-call NopRetryer; DirStore's is Put; prefixStore forwards). moves gains the create and place kinds, epoch and rev, HostKey and the entry rules, PutTable and TableWrites (Unanswered, Resend), ErrUncertain, PeekActive, ReadActive, ReadForAcquire, NextActiveBody, Holds and Release, the frozen plan 3 parser as a test witness, and the table-write audit, which keeps the console's three plan 2 sites by an explicit list until Task 2b.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
9 files changed,
+1760
−13
internal/moves/offline.go+65 −12internal/moves/offline_compat_test.go+173 −0internal/moves/reserve.go+196 −0internal/moves/reserve_test.go+542 −0internal/moves/tablewrite.go+172 −0internal/moves/tablewrite_audit_test.go+211 −0internal/moves/tablewrite_test.go+171 −0internal/release/putonce_test.go+174 −0internal/release/store.go+56 −1
diff --git a/internal/moves/offline.go b/internal/moves/offline.go| index 733c38610..16eac114d 100644 |
| --- a/internal/moves/offline.go |
| +++ b/internal/moves/offline.go |
| @@ -533,7 +533,13 @@ func OfflineStager(boxes []string) string { |
| const OfflineActiveKey = "control/offline-moves/active.json" |
| // OfflineActive is active.json's body. Reservations is the table, keyed by |
| -// box id (OfflineSentinel for an operation that reserves every box). Host, |
| +// box id (OfflineSentinel for an operation that reserves every box, HostKey |
| +// for an operation's hold on a hostname). Epoch and Rev make every body the |
| +// table holds distinct (plan 4, Global Constraint "The table is never |
| +// deleted"): the first writer creates {v: 1, epoch, rev: 1}, and every later |
| +// write carries the read body's epoch and its rev plus one (NextActiveBody), |
| +// so no ETag ever comes back. A console before plan 4 writes neither; its |
| +// body is normalised before any acquisition (ReadForAcquire). Host, |
| // Op, Account, App and At are the projection: the oldest move entry's host, |
| // id, account, app and reserved_at (the oldest entry of any kind when no |
| // move holds a box), so a console that predates the table reads one lock |
| @@ -541,6 +547,8 @@ const OfflineActiveKey = "control/offline-moves/active.json" |
| // no Reservations and V 0 is a legacy lock (ParseOfflineActive). |
| type OfflineActive struct { |
| V int `json:"v,omitempty"` |
| + Epoch string `json:"epoch,omitempty"` |
| + Rev uint64 `json:"rev,omitempty"` |
| Host string `json:"host"` |
| Op string `json:"op"` |
| Account string `json:"account"` |
| @@ -559,6 +567,13 @@ const ( |
| // OfflineKindCleanup is a box's cleanup of a retired host (plan 4); |
| // ID is its retired.json's id. |
| OfflineKindCleanup = "cleanup" |
| + // OfflineKindCreate is the console's create of an instance (plan 4), |
| + // holding its hostname's key; ID is the allocation's op id. |
| + OfflineKindCreate = "create" |
| + // OfflineKindPlace is the console's placement of an app's instances |
| + // (plan 4), holding every instance's hostname key; ID is the |
| + // placement's id. |
| + OfflineKindPlace = "place" |
| // OfflineSentinel is the key of an operation that reserves every box: |
| // a selection with a participant whose InstanceID is empty, and a |
| // legacy lock. It conflicts with every other acquisition, and every |
| @@ -580,9 +595,39 @@ type OfflineReservation struct { |
| ExpectIncarnation uint64 `json:"expect_incarnation,omitempty"` |
| } |
| -// valid: a known kind, an id, a host and a time. |
| +// hostKeyPrefix starts every hostname key. A box label never contains "/" |
| +// (controlname.ValidLabel), so no box key is a host key. |
| +const hostKeyPrefix = "host/" |
| + |
| +// HostKey is the reservation key of an operation's hold on host: |
| +// "host/" + host. |
| +func HostKey(host string) string { return hostKeyPrefix + host } |
| + |
| +// IsHostKey reports whether key is a HostKey. |
| +func IsHostKey(key string) bool { return strings.HasPrefix(key, hostKeyPrefix) } |
| + |
| +// valid: a known kind, an id, a host and a time; a create or a place also |
| +// names its account and app. |
| func (r OfflineReservation) valid() bool { |
| - return (r.Kind == OfflineKindMove || r.Kind == OfflineKindCleanup) && r.ID != "" && r.Host != "" && !r.ReservedAt.IsZero() |
| + switch r.Kind { |
| + case OfflineKindMove, OfflineKindCleanup: |
| + case OfflineKindCreate, OfflineKindPlace: |
| + if r.Account == "" || r.App == "" { |
| + return false |
| + } |
| + default: |
| + return false |
| + } |
| + return r.ID != "" && r.Host != "" && !r.ReservedAt.IsZero() |
| +} |
| + |
| +// validAt: valid, under a non-empty key, and an entry under a host key |
| +// names that host. |
| +func (r OfflineReservation) validAt(key string) bool { |
| + if key == "" || !r.valid() { |
| + return false |
| + } |
| + return !IsHostKey(key) || r.Host == strings.TrimPrefix(key, hostKeyPrefix) |
| } |
| // ErrOfflineActiveMalformed is a body that is not a table and not a legacy |
| @@ -594,31 +639,39 @@ var ErrOfflineActiveMalformed = errors.New("moves: active.json is malformed") |
| // ParseOfflineActive reads active.json's body as its reservation table. A |
| // legacy lock (no Reservations, V 0) reads as the one sentinel entry |
| // {move, Op, Host, Account, App, At}. An unknown version, a table without |
| -// V, an empty key, or an entry with an unknown kind, an empty id or host, or |
| -// a zero time is ErrOfflineActiveMalformed. |
| +// V, an empty key, or an entry with an unknown kind, an empty id or host, a |
| +// zero time, a create or place without its account or app, or an entry |
| +// under a host key naming another host is ErrOfflineActiveMalformed. A body |
| +// without epoch or rev (a console before plan 4 wrote it) parses. |
| func ParseOfflineActive(raw []byte) (map[string]OfflineReservation, error) { |
| + _, table, err := parseOfflineActiveBody(raw) |
| + return table, err |
| +} |
| + |
| +// parseOfflineActiveBody is ParseOfflineActive with the body it read. |
| +func parseOfflineActiveBody(raw []byte) (OfflineActive, map[string]OfflineReservation, error) { |
| var a OfflineActive |
| if err := json.Unmarshal(raw, &a); err != nil { |
| - return nil, fmt.Errorf("%w: %v", ErrOfflineActiveMalformed, err) |
| + return a, nil, fmt.Errorf("%w: %v", ErrOfflineActiveMalformed, err) |
| } |
| if a.V == 0 && a.Reservations == nil { |
| e := OfflineReservation{Kind: OfflineKindMove, ID: a.Op, Host: a.Host, Account: a.Account, App: a.App, ReservedAt: a.At} |
| if !e.valid() { |
| - return nil, fmt.Errorf("%w: a legacy lock without a host, an op or a time", ErrOfflineActiveMalformed) |
| + return a, nil, fmt.Errorf("%w: a legacy lock without a host, an op or a time", ErrOfflineActiveMalformed) |
| } |
| - return map[string]OfflineReservation{OfflineSentinel: e}, nil |
| + return a, map[string]OfflineReservation{OfflineSentinel: e}, nil |
| } |
| if a.V != OfflineActiveVersion { |
| - return nil, fmt.Errorf("%w: version %d", ErrOfflineActiveMalformed, a.V) |
| + return a, nil, fmt.Errorf("%w: version %d", ErrOfflineActiveMalformed, a.V) |
| } |
| table := make(map[string]OfflineReservation, len(a.Reservations)) |
| for k, e := range a.Reservations { |
| - if k == "" || !e.valid() { |
| - return nil, fmt.Errorf("%w: entry %q", ErrOfflineActiveMalformed, k) |
| + if !e.validAt(k) { |
| + return a, nil, fmt.Errorf("%w: entry %q", ErrOfflineActiveMalformed, k) |
| } |
| table[k] = e |
| } |
| - return table, nil |
| + return a, table, nil |
| } |
| // OfflineActiveBody is the table form of table: V, the table, and the |
diff --git a/internal/moves/offline_compat_test.go b/internal/moves/offline_compat_test.go| new file mode 100644 |
| index 000000000..5b5afda96 |
| --- /dev/null |
| +++ b/internal/moves/offline_compat_test.go |
| @@ -0,0 +1,173 @@ |
| +package moves |
| + |
| +import ( |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "strings" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// plan3OfflineActive and plan3OfflineReservation are active.json's body and |
| +// entry as a console before plan 4 declares them: no epoch, no rev. |
| +type plan3OfflineActive struct { |
| + V int `json:"v,omitempty"` |
| + Host string `json:"host"` |
| + Op string `json:"op"` |
| + Account string `json:"account"` |
| + App string `json:"app"` |
| + At time.Time `json:"at"` |
| + Reservations map[string]plan3OfflineReservation `json:"reservations,omitempty"` |
| +} |
| + |
| +type plan3OfflineReservation struct { |
| + Kind string `json:"kind"` |
| + ID string `json:"id"` |
| + Host string `json:"host"` |
| + Account string `json:"account"` |
| + App string `json:"app"` |
| + ReservedAt time.Time `json:"reserved_at"` |
| + ExpectSeq uint64 `json:"expect_seq,omitempty"` |
| + ExpectIncarnation uint64 `json:"expect_incarnation,omitempty"` |
| +} |
| + |
| +// plan3Valid and parsePlan3OfflineActive are a verbatim copy, frozen, of |
| +// OfflineReservation.valid() and ParseOfflineActive as built at 0840937f |
| +// (internal/moves/offline.go:584-587 and :599-622), over the plan 3 types |
| +// above. They are the witness of what a console from before plan 4 reads: |
| +// a create or place entry is a malformed table to it, which is why no such |
| +// console may run once a plan 4 console has written the table (Global |
| +// Constraint "No console rollback across the plan 4 boundary"). Never edit |
| +// them to follow the live parser. |
| +func (r plan3OfflineReservation) plan3Valid() bool { |
| + return (r.Kind == OfflineKindMove || r.Kind == OfflineKindCleanup) && r.ID != "" && r.Host != "" && !r.ReservedAt.IsZero() |
| +} |
| + |
| +func parsePlan3OfflineActive(raw []byte) (map[string]plan3OfflineReservation, error) { |
| + var a plan3OfflineActive |
| + if err := json.Unmarshal(raw, &a); err != nil { |
| + return nil, fmt.Errorf("%w: %v", ErrOfflineActiveMalformed, err) |
| + } |
| + if a.V == 0 && a.Reservations == nil { |
| + e := plan3OfflineReservation{Kind: OfflineKindMove, ID: a.Op, Host: a.Host, Account: a.Account, App: a.App, ReservedAt: a.At} |
| + if !e.plan3Valid() { |
| + return nil, fmt.Errorf("%w: a legacy lock without a host, an op or a time", ErrOfflineActiveMalformed) |
| + } |
| + return map[string]plan3OfflineReservation{OfflineSentinel: e}, nil |
| + } |
| + if a.V != OfflineActiveVersion { |
| + return nil, fmt.Errorf("%w: version %d", ErrOfflineActiveMalformed, a.V) |
| + } |
| + table := make(map[string]plan3OfflineReservation, len(a.Reservations)) |
| + for k, e := range a.Reservations { |
| + if k == "" || !e.plan3Valid() { |
| + return nil, fmt.Errorf("%w: entry %q", ErrOfflineActiveMalformed, k) |
| + } |
| + table[k] = e |
| + } |
| + return table, nil |
| +} |
| + |
| +var compatAt = time.Date(2026, 10, 24, 9, 0, 0, 0, time.UTC) |
| + |
| +func entryJSON(key, kind, id, host, acct, app string, at bool) string { |
| + e := map[string]any{"kind": kind, "id": id, "host": host, "account": acct, "app": app} |
| + if at { |
| + e["reserved_at"] = compatAt |
| + } |
| + b, _ := json.Marshal(map[string]any{"v": 1, "epoch": "00112233445566778899aabbccddeeff", "rev": 3, "reservations": map[string]any{key: e}}) |
| + return string(b) |
| +} |
| + |
| +// The entry rules (Global Constraint "Reservation keys and the one |
| +// conflict relation"): a known kind, a non-empty id and host, a non-zero |
| +// reserved_at; for create and place a non-empty account and app; an entry |
| +// under a host key names that host. |
| +func TestEntryShapeRules(t *testing.T) { |
| + valid := map[string]string{ |
| + "move on a box": entryJSON("i-0a", OfflineKindMove, "mv_a", "shop.example.com", "acct", "shop", true), |
| + "move on the sentinel": entryJSON(OfflineSentinel, OfflineKindMove, "mv_a", "shop.example.com", "acct", "shop", true), |
| + "move on its host key": entryJSON(HostKey("shop.example.com"), OfflineKindMove, "mv_a", "shop.example.com", "acct", "shop", true), |
| + "cleanup without an app": entryJSON("i-0a", OfflineKindCleanup, "cu_1", "old.example.com", "", "", true), |
| + "cleanup on its host key": entryJSON(HostKey("old.example.com"), OfflineKindCleanup, "cu_1", "old.example.com", "", "", true), |
| + "create on its host key": entryJSON(HostKey("new.example.com"), OfflineKindCreate, "op_1", "new.example.com", "acct", "shop", true), |
| + "place on its host key": entryJSON(HostKey("a.example.com"), OfflineKindPlace, "pl_1", "a.example.com", "acct", "shop", true), |
| + "an empty plan 4 table": `{"v":1,"epoch":"00112233445566778899aabbccddeeff","rev":1}`, |
| + "a plan 3 table, no epoch": `{"v":1,"reservations":{"i-0a":{"kind":"move","id":"mv_a","host":"a","reserved_at":"2026-10-22T11:00:00Z"}}}`, |
| + } |
| + for name, raw := range valid { |
| + if _, err := ParseOfflineActive([]byte(raw)); err != nil { |
| + t.Errorf("%s: %v", name, err) |
| + } |
| + } |
| + malformed := map[string]string{ |
| + "create without an account": entryJSON(HostKey("new.example.com"), OfflineKindCreate, "op_1", "new.example.com", "", "shop", true), |
| + "create without an app": entryJSON(HostKey("new.example.com"), OfflineKindCreate, "op_1", "new.example.com", "acct", "", true), |
| + "place without an account": entryJSON(HostKey("a.example.com"), OfflineKindPlace, "pl_1", "a.example.com", "", "shop", true), |
| + "place without an app": entryJSON(HostKey("a.example.com"), OfflineKindPlace, "pl_1", "a.example.com", "acct", "", true), |
| + "host key naming another": entryJSON(HostKey("b.example.com"), OfflineKindPlace, "pl_1", "a.example.com", "acct", "shop", true), |
| + "move on another's host": entryJSON(HostKey("b.example.com"), OfflineKindMove, "mv_a", "a.example.com", "acct", "shop", true), |
| + "an empty host key": entryJSON(HostKey(""), OfflineKindCleanup, "cu_1", "a.example.com", "", "", true), |
| + "an unknown kind": entryJSON("i-0a", "lease", "x", "a.example.com", "acct", "shop", true), |
| + "a create without a time": entryJSON(HostKey("new.example.com"), OfflineKindCreate, "op_1", "new.example.com", "acct", "shop", false), |
| + "a place without an id": entryJSON(HostKey("a.example.com"), OfflineKindPlace, "", "a.example.com", "acct", "shop", true), |
| + } |
| + for name, raw := range malformed { |
| + if got, err := ParseOfflineActive([]byte(raw)); !errors.Is(err, ErrOfflineActiveMalformed) { |
| + t.Errorf("%s: parsed as %+v %v", name, got, err) |
| + } |
| + } |
| + if HostKey("shop.example.com") != "host/shop.example.com" || !IsHostKey(HostKey("shop.example.com")) || IsHostKey("i-0a") || IsHostKey(OfflineSentinel) { |
| + t.Fatal("HostKey and IsHostKey disagree") |
| + } |
| +} |
| + |
| +// The new parser reads every table plan 3 wrote; a body NextActiveBody |
| +// wrote with a create or a place entry parses with it, and the frozen plan 3 |
| +// parser answers it malformed, which pins the reason for the no-rollback |
| +// rule. |
| +func TestNewParserReadsThePlan3Table(t *testing.T) { |
| + move := plan3OfflineReservation{Kind: OfflineKindMove, ID: "mv_a", Host: "a.example.com", Account: "acct", App: "shop", ReservedAt: compatAt, ExpectSeq: 4, ExpectIncarnation: 2} |
| + cleanup := plan3OfflineReservation{Kind: OfflineKindCleanup, ID: "cu_1", Host: "old.example.com", ReservedAt: compatAt.Add(time.Minute)} |
| + plan3 := map[string]plan3OfflineActive{ |
| + "moves": {V: 1, Host: "a.example.com", Op: "mv_a", Account: "acct", App: "shop", At: compatAt, Reservations: map[string]plan3OfflineReservation{"i-0a": move, "i-0b": move, "i-0c": cleanup}}, |
| + "sentinel": {V: 1, Host: "a.example.com", Op: "mv_a", Account: "acct", App: "shop", At: compatAt, Reservations: map[string]plan3OfflineReservation{OfflineSentinel: move}}, |
| + "legacy": {Host: "a.example.com", Op: "mv_a", Account: "acct", App: "shop", At: compatAt}, |
| + } |
| + for name, body := range plan3 { |
| + raw, _ := json.Marshal(body) |
| + got, err := ParseOfflineActive(raw) |
| + if err != nil { |
| + t.Fatalf("%s: %v", name, err) |
| + } |
| + old, err := parsePlan3OfflineActive(raw) |
| + if err != nil || len(old) != len(got) { |
| + t.Fatalf("%s: the plan 3 parser read %d entries (%v), the new one %d", name, len(old), err, len(got)) |
| + } |
| + for k, e := range old { |
| + n := got[k] |
| + if n.Kind != e.Kind || n.ID != e.ID || n.Host != e.Host || n.Account != e.Account || n.App != e.App || !n.ReservedAt.Equal(e.ReservedAt) || n.ExpectSeq != e.ExpectSeq || n.ExpectIncarnation != e.ExpectIncarnation { |
| + t.Fatalf("%s: entry %s read as %+v, plan 3 read %+v", name, k, n, e) |
| + } |
| + } |
| + } |
| + |
| + for _, kind := range []string{OfflineKindCreate, OfflineKindPlace} { |
| + table := map[string]OfflineReservation{ |
| + "i-0a": {Kind: OfflineKindMove, ID: "mv_a", Host: "a.example.com", Account: "acct", App: "shop", ReservedAt: compatAt}, |
| + HostKey("n.example.com"): {Kind: kind, ID: "op_" + kind, Host: "n.example.com", Account: "acct", App: "shop", ReservedAt: compatAt}, |
| + } |
| + raw, err := json.Marshal(NextActiveBody(OfflineActive{V: 1, Epoch: "00112233445566778899aabbccddeeff", Rev: 7}, table)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + got, err := ParseOfflineActive(raw) |
| + if err != nil || len(got) != 2 || got[HostKey("n.example.com")].Kind != kind { |
| + t.Fatalf("%s: the new parser read %+v %v", kind, got, err) |
| + } |
| + if old, err := parsePlan3OfflineActive(raw); !errors.Is(err, ErrOfflineActiveMalformed) || !strings.Contains(err.Error(), "host/n.example.com") { |
| + t.Fatalf("%s: the plan 3 parser read %+v %v, want the table malformed", kind, old, err) |
| + } |
| + } |
| +} |
diff --git a/internal/moves/reserve.go b/internal/moves/reserve.go| new file mode 100644 |
| index 000000000..2dc5a54a2 |
| --- /dev/null |
| +++ b/internal/moves/reserve.go |
| @@ -0,0 +1,196 @@ |
| +package moves |
| + |
| +import ( |
| + "context" |
| + "crypto/rand" |
| + "encoding/hex" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "os" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// The reservation table's readers and its release (plan 4, Task 2a). The |
| +// table is created once, by the first writer, as {v: 1, epoch, rev: 1} |
| +// create-only, and never deleted: emptiness is a table with no entries. |
| +// Every later write is one PutTable with If-Match the ETag it read, of |
| +// NextActiveBody, so no two bodies the table holds are equal and an ETag |
| +// never comes back. A conditional write that read ETag E can land only |
| +// while E is current, so once any later write succeeds every earlier |
| +// outstanding write that read E can never land: an uncertain acquisition is |
| +// settled by the next successful conditional write by anyone. |
| +// |
| +// Three readers. PeekActive only reads (every box read, root's authority |
| +// read). ReadActive creates an absent table first (the armed writers). |
| +// ReadForAcquire also normalises a body without epoch or rev before an |
| +// acquisition may name its ETag (T4). |
| + |
| +// Op is one operation's identity in the table: its kind and id. |
| +type Op struct{ Kind, ID string } |
| + |
| +// opOf is e's operation. |
| +func opOf(e OfflineReservation) Op { return Op{Kind: e.Kind, ID: e.ID} } |
| + |
| +// ErrUnreadable is a table that could not be read: a store error, a |
| +// malformed body (ErrOfflineActiveMalformed, wrapped), or a body that could |
| +// not be normalised. Nothing is acquired or released against it. |
| +var ErrUnreadable = errors.New("moves: the reservation table could not be read") |
| + |
| +// Active is the table as one read saw it: its body, its entries, and the |
| +// ETag a write against it names ("" for an absent table). |
| +type Active struct { |
| + Body OfflineActive |
| + Table map[string]OfflineReservation |
| + ETag string |
| +} |
| + |
| +// newEpoch is 16 random bytes, hex. |
| +func newEpoch() string { |
| + var b [16]byte |
| + if _, err := rand.Read(b[:]); err != nil { |
| + panic("moves: no randomness for the table's epoch: " + err.Error()) |
| + } |
| + return hex.EncodeToString(b[:]) |
| +} |
| + |
| +// getActive reads the table once: found false for an absent table. |
| +func getActive(ctx context.Context, s release.Store) (Active, bool, error) { |
| + raw, etag, err := s.Get(ctx, OfflineActiveKey) |
| + if errors.Is(err, os.ErrNotExist) { |
| + return Active{Table: map[string]OfflineReservation{}}, false, nil |
| + } |
| + if err != nil { |
| + return Active{}, false, fmt.Errorf("%w: %w", ErrUnreadable, err) |
| + } |
| + body, table, err := parseOfflineActiveBody(raw) |
| + if err != nil { |
| + return Active{}, false, fmt.Errorf("%w: %w", ErrUnreadable, err) |
| + } |
| + return Active{Body: body, Table: table, ETag: etag}, true, nil |
| +} |
| + |
| +// PeekActive reads the table and never writes it (V3): an absent table is |
| +// an empty table with ETag "", and a metadata-free body is answered as read. |
| +// A read error or a malformed body is ErrUnreadable. |
| +func PeekActive(ctx context.Context, s release.Store) (Active, error) { |
| + a, _, err := getActive(ctx, s) |
| + return a, err |
| +} |
| + |
| +// ReadActive reads the table, creating it first when it is absent: one |
| +// create-only PutTable of {v: 1, epoch, rev: 1}, then a second read whatever |
| +// that write answered (a lost create means another writer created it; an |
| +// uncertain one can only ever create an empty table). A store without |
| +// PutOnce answers release.ErrNoSingleAttemptPut for an absent table, with |
| +// nothing sent. A read error or a malformed body is ErrUnreadable. Only an |
| +// armed writer reaches this on a box; every other read is PeekActive. |
| +func ReadActive(ctx context.Context, s release.Store, tw *TableWrites) (Active, error) { |
| + a, found, err := getActive(ctx, s) |
| + if err != nil || found { |
| + return a, err |
| + } |
| + raw, err := json.Marshal(OfflineActive{V: OfflineActiveVersion, Epoch: newEpoch(), Rev: 1}) |
| + if err != nil { |
| + return Active{}, err |
| + } |
| + if _, err := PutTable(ctx, s, tw, raw, ""); errors.Is(err, release.ErrNoSingleAttemptPut) { |
| + return Active{}, err |
| + } |
| + a, found, err = getActive(ctx, s) |
| + if err == nil && !found { |
| + err = fmt.Errorf("%w: absent after its creation", ErrUnreadable) |
| + } |
| + return a, err |
| +} |
| + |
| +// NextActiveBody is the body a write against prev puts: table's body |
| +// (OfflineActiveBody) with prev's epoch (a new one when prev has none, as |
| +// a console before plan 4 wrote) and prev's rev plus one. |
| +func NextActiveBody(prev OfflineActive, table map[string]OfflineReservation) OfflineActive { |
| + next := OfflineActiveBody(table) |
| + next.Epoch = prev.Epoch |
| + if next.Epoch == "" { |
| + next.Epoch = newEpoch() |
| + } |
| + next.Rev = prev.Rev + 1 |
| + return next |
| +} |
| + |
| +// normalised: the body carries an epoch and a rev, so its ETag can never |
| +// come back. |
| +func normalised(b OfflineActive) bool { return b.Epoch != "" && b.Rev != 0 } |
| + |
| +// ReadForAcquire is the read of every acquisition (T4): ReadActive, and a |
| +// body without epoch or rev (a console before plan 4 wrote it) is first |
| +// normalised by one PutTable, If-Match the read ETag, of NextActiveBody (the |
| +// same entries, a fresh epoch, rev 1), whatever it answers, then read |
| +// again; at most three rounds. So only an epoch-bearing ETag is ever an |
| +// acquisition's condition. A body still without both is ErrUnreadable. |
| +func ReadForAcquire(ctx context.Context, s release.Store, tw *TableWrites) (Active, error) { |
| + for round := 0; ; round++ { |
| + a, err := ReadActive(ctx, s, tw) |
| + if err != nil || normalised(a.Body) { |
| + return a, err |
| + } |
| + if round == 3 { |
| + return Active{}, fmt.Errorf("%w: the table could not be normalised", ErrUnreadable) |
| + } |
| + raw, err := json.Marshal(NextActiveBody(a.Body, a.Table)) |
| + if err != nil { |
| + return Active{}, err |
| + } |
| + if _, err := PutTable(ctx, s, tw, raw, a.ETag); errors.Is(err, release.ErrNoSingleAttemptPut) { |
| + return Active{}, err |
| + } |
| + } |
| +} |
| + |
| +// Holds reports whether mine holds every one of keys in table. No keys is |
| +// false: an operation is never said to hold nothing. |
| +func Holds(table map[string]OfflineReservation, keys []string, mine Op) bool { |
| + if len(keys) == 0 { |
| + return false |
| + } |
| + for _, k := range keys { |
| + e, ok := table[k] |
| + if !ok || opOf(e) != mine { |
| + return false |
| + } |
| + } |
| + return true |
| +} |
| + |
| +// Release removes every entry of mine and always writes: one PutTable, If-Match |
| +// the read ETag, of NextActiveBody, even when mine held nothing. Its own |
| +// success is the settlement its caller needs: no earlier write of mine that |
| +// read an older ETag can land after it, which a read alone never proves. A |
| +// lost condition reads again, three rounds; a malformed table is an error |
| +// and is left as it is. Any kind of operation may be released. |
| +func Release(ctx context.Context, s release.Store, tw *TableWrites, mine Op) error { |
| + for round := 0; round < 3; round++ { |
| + a, err := ReadActive(ctx, s, tw) |
| + if err != nil { |
| + return fmt.Errorf("moves: releasing %s %s: %w", mine.Kind, mine.ID, err) |
| + } |
| + next := make(map[string]OfflineReservation, len(a.Table)) |
| + for k, e := range a.Table { |
| + if opOf(e) != mine { |
| + next[k] = e |
| + } |
| + } |
| + raw, err := json.Marshal(NextActiveBody(a.Body, next)) |
| + if err != nil { |
| + return err |
| + } |
| + _, err = PutTable(ctx, s, tw, raw, a.ETag) |
| + if err == nil { |
| + return nil |
| + } |
| + if !errors.Is(err, release.ErrCASConflict) { |
| + return fmt.Errorf("moves: releasing %s %s: %w", mine.Kind, mine.ID, err) |
| + } |
| + } |
| + return fmt.Errorf("moves: releasing %s %s: lost 3 rounds", mine.Kind, mine.ID) |
| +} |
diff --git a/internal/moves/reserve_test.go b/internal/moves/reserve_test.go| new file mode 100644 |
| index 000000000..7542ee789 |
| --- /dev/null |
| +++ b/internal/moves/reserve_test.go |
| @@ -0,0 +1,542 @@ |
| +package moves |
| + |
| +import ( |
| + "context" |
| + "crypto/sha256" |
| + "encoding/hex" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "net/http" |
| + "net/http/httptest" |
| + "os" |
| + "strings" |
| + "sync" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/fleetapi" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// tableStore is an in-memory deployment bucket whose ETag is the body's |
| +// hash, as S3's is for a single-part object, so an ETag recurs exactly when |
| +// a body does. It logs every call ("Get", "Put(<ifMatch>)", "PutOnce(<ifMatch>)", |
| +// "Delete", "DeleteIfMatch") and lets a test intercept each PutOnce. |
| +type tableStore struct { |
| + mu sync.Mutex |
| + objects map[string][]byte |
| + log []string |
| + sent []sentPut |
| + // onPutOnce, when set, runs before a PutOnce is applied, outside the |
| + // lock (it may block): commit says whether the write is applied, err |
| + // what the caller is answered (after the commit, when both). |
| + onPutOnce func(n int, key string, body []byte, ifMatch string) (commit bool, err error) |
| + // onGet, when set, runs after every Get has read, outside the lock. |
| + onGet func(n int) |
| + gets int |
| +} |
| + |
| +type sentPut struct { |
| + key, ifMatch string |
| + body []byte |
| +} |
| + |
| +func newTableStore() *tableStore { return &tableStore{objects: map[string][]byte{}} } |
| + |
| +func etagFor(b []byte) string { |
| + sum := sha256.Sum256(b) |
| + return `"` + hex.EncodeToString(sum[:8]) + `"` |
| +} |
| + |
| +func (f *tableStore) record(s string) { |
| + f.mu.Lock() |
| + f.log = append(f.log, s) |
| + f.mu.Unlock() |
| +} |
| + |
| +func (f *tableStore) Log() []string { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + return append([]string(nil), f.log...) |
| +} |
| + |
| +func (f *tableStore) Requests() int { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + return len(f.sent) |
| +} |
| + |
| +func (f *tableStore) Sent() []sentPut { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + return append([]sentPut(nil), f.sent...) |
| +} |
| + |
| +// set writes key's body directly, as another writer would. |
| +func (f *tableStore) set(key string, body []byte) string { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + f.objects[key] = append([]byte(nil), body...) |
| + return etagFor(body) |
| +} |
| + |
| +func (f *tableStore) current(key string) ([]byte, string, bool) { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + b, ok := f.objects[key] |
| + return b, etagFor(b), ok |
| +} |
| + |
| +func (f *tableStore) Get(_ context.Context, key string) ([]byte, string, error) { |
| + f.mu.Lock() |
| + n := f.gets |
| + f.gets++ |
| + hook := f.onGet |
| + b, ok := f.objects[key] |
| + b = append([]byte(nil), b...) |
| + f.log = append(f.log, "Get") |
| + f.mu.Unlock() |
| + if hook != nil { |
| + hook(n) // after the read: what it answers is already decided |
| + } |
| + if !ok { |
| + return nil, "", fmt.Errorf("fake: %s: %w", key, os.ErrNotExist) |
| + } |
| + return b, etagFor(b), nil |
| +} |
| + |
| +func (f *tableStore) apply(key string, body []byte, ifMatch string) (string, error) { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + cur, ok := f.objects[key] |
| + switch { |
| + case ifMatch == "" && ok: |
| + return "", release.ErrCASConflict |
| + case ifMatch != "" && (!ok || etagFor(cur) != ifMatch): |
| + return "", release.ErrCASConflict |
| + } |
| + f.objects[key] = append([]byte(nil), body...) |
| + return etagFor(body), nil |
| +} |
| + |
| +func (f *tableStore) Put(_ context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + f.record("Put(" + ifMatch + ")") |
| + return f.apply(key, body, ifMatch) |
| +} |
| + |
| +func (f *tableStore) PutOnce(_ context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + f.mu.Lock() |
| + n := len(f.sent) |
| + f.sent = append(f.sent, sentPut{key, ifMatch, append([]byte(nil), body...)}) |
| + hook := f.onPutOnce |
| + f.mu.Unlock() |
| + f.record("PutOnce(" + ifMatch + ")") |
| + commit, err := true, error(nil) |
| + if hook != nil { |
| + commit, err = hook(n, key, body, ifMatch) |
| + } |
| + var etag string |
| + var cerr error |
| + if commit { |
| + etag, cerr = f.apply(key, body, ifMatch) |
| + } |
| + if err != nil { |
| + return "", err |
| + } |
| + return etag, cerr |
| +} |
| + |
| +func (f *tableStore) List(context.Context, string) ([]string, error) { return nil, nil } |
| +func (f *tableStore) ListDirs(context.Context, string) ([]string, error) { return nil, nil } |
| +func (f *tableStore) Delete(_ context.Context, key string) error { |
| + f.record("Delete") |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + delete(f.objects, key) |
| + return nil |
| +} |
| +func (f *tableStore) DeleteIfMatch(_ context.Context, key, etag string) error { |
| + f.record("DeleteIfMatch") |
| + return errors.New("fake: no conditional deletes of the table") |
| +} |
| + |
| +// errTimeout is an attempt that was not answered. |
| +var errTimeout = errors.New("fake: request timed out") |
| + |
| +func fastResend(t *testing.T) { |
| + t.Helper() |
| + t.Cleanup(SetTableRetryScheduleForTest(time.Millisecond, time.Millisecond)) |
| +} |
| + |
| +func tableBody(t *testing.T, a OfflineActive) []byte { |
| + t.Helper() |
| + b, err := json.Marshal(a) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return b |
| +} |
| + |
| +func readBody(t *testing.T, f *tableStore) OfflineActive { |
| + t.Helper() |
| + b, _, ok := f.current(OfflineActiveKey) |
| + if !ok { |
| + t.Fatal("the table is absent") |
| + } |
| + var a OfflineActive |
| + if err := json.Unmarshal(b, &a); err != nil { |
| + t.Fatal(err) |
| + } |
| + return a |
| +} |
| + |
| +func moveEntry(id, host string) OfflineReservation { |
| + return OfflineReservation{Kind: OfflineKindMove, ID: id, Host: host, Account: "acct", App: "shop", ReservedAt: compatAt} |
| +} |
| + |
| +// plan3Body is a table as a console before plan 4 writes it: v 1, entries, |
| +// the projection, no epoch and no rev. |
| +func plan3Body(t *testing.T, table map[string]OfflineReservation) []byte { |
| + t.Helper() |
| + b := tableBody(t, OfflineActiveBody(table)) |
| + if strings.Contains(string(b), "epoch") || strings.Contains(string(b), `"rev"`) { |
| + t.Fatalf("a plan 3 body carries %s", b) |
| + } |
| + return b |
| +} |
| + |
| +// acquire is an acquisition's write as Task 2b's Reserve makes it: one |
| +// PutTable of NextActiveBody with the entries added, If-Match the read ETag. |
| +func acquire(ctx context.Context, t *testing.T, s release.Store, tw *TableWrites, a Active, add map[string]OfflineReservation) (string, error) { |
| + t.Helper() |
| + next := map[string]OfflineReservation{} |
| + for k, e := range a.Table { |
| + next[k] = e |
| + } |
| + for k, e := range add { |
| + next[k] = e |
| + } |
| + return PutTable(ctx, s, tw, tableBody(t, NextActiveBody(a.Body, next)), a.ETag) |
| +} |
| + |
| +func TestActiveTableIsNeverDeleted(t *testing.T) { |
| + ctx := context.Background() |
| + f := newTableStore() |
| + var tw TableWrites |
| + a, err := ReadActive(ctx, f, &tw) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := acquire(ctx, t, f, &tw, a, map[string]OfflineReservation{"i-0a": moveEntry("mv_a", "a.example.com"), HostKey("a.example.com"): moveEntry("mv_a", "a.example.com")}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := Release(ctx, f, &tw, Op{OfflineKindMove, "mv_a"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + body := readBody(t, f) |
| + if len(body.Reservations) != 0 || body.V != OfflineActiveVersion || body.Epoch == "" || body.Rev != 3 { |
| + t.Fatalf("after the last release the table is %+v, want an empty table at rev 3", body) |
| + } |
| + for _, l := range f.Log() { |
| + if l == "Delete" || l == "DeleteIfMatch" || strings.HasPrefix(l, "Put(") { |
| + t.Fatalf("the table saw %s: %q", l, f.Log()) |
| + } |
| + } |
| +} |
| + |
| +// Two first writers: one bootstrap lands, create-only and empty, and both |
| +// then write through If-Match. |
| +func TestBootstrapIsCreateOnlyAndEmpty(t *testing.T) { |
| + ctx := context.Background() |
| + f := newTableStore() |
| + var both sync.WaitGroup |
| + both.Add(2) |
| + f.onGet = func(n int) { |
| + if n < 2 { // both first reads see the table absent before either creates it |
| + both.Done() |
| + both.Wait() |
| + } |
| + } |
| + var tw TableWrites |
| + errs := make(chan error, 2) |
| + for _, id := range []string{"mv_a", "mv_b"} { |
| + go func(id string) { errs <- Release(ctx, f, &tw, Op{OfflineKindMove, id}) }(id) |
| + } |
| + for range 2 { |
| + if err := <-errs; err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + var creates int |
| + for i, p := range f.Sent() { |
| + if p.ifMatch != "" { |
| + continue |
| + } |
| + creates++ |
| + var a OfflineActive |
| + if err := json.Unmarshal(p.body, &a); err != nil { |
| + t.Fatal(err) |
| + } |
| + if a.V != OfflineActiveVersion || a.Rev != 1 || len(a.Epoch) != 32 || len(a.Reservations) != 0 { |
| + t.Fatalf("bootstrap %d is %s", i, p.body) |
| + } |
| + if _, err := hex.DecodeString(a.Epoch); err != nil { |
| + t.Fatalf("epoch %q: %v", a.Epoch, err) |
| + } |
| + } |
| + // Each writer's bootstrap is sent once; one landed (the table's rev |
| + // counts one create and the two settling writes). |
| + if body := readBody(t, f); body.Rev != 3 { |
| + t.Fatalf("the table is at rev %d, want 3 (one create, two releases)", body.Rev) |
| + } |
| + if creates != 2 { |
| + t.Fatalf("%d create-only writes were sent, want 2 (one lands): %q", creates, f.Log()) |
| + } |
| + if n := len(f.Sent()) - creates; n < 2 { |
| + t.Fatalf("only %d writes through If-Match: both writers did not write after the bootstrap: %q", n, f.Log()) |
| + } |
| +} |
| + |
| +// Alternating writes of one entry never repeat a body, so never an ETag; |
| +// a body without epoch, as plan 3 writes, gets one. |
| +func TestRevAndEpochNeverRepeatABody(t *testing.T) { |
| + ctx := context.Background() |
| + f := newTableStore() |
| + var tw TableWrites |
| + seen := map[string]int{} |
| + note := func(etag string) { |
| + t.Helper() |
| + seen[etag]++ |
| + if seen[etag] > 1 { |
| + t.Fatalf("ETag %s recurred", etag) |
| + } |
| + } |
| + for i := range 4 { |
| + a, err := ReadActive(ctx, f, &tw) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if i == 0 { |
| + note(a.ETag) |
| + } |
| + etag, err := acquire(ctx, t, f, &tw, a, map[string]OfflineReservation{"i-0a": moveEntry("mv_a", "a.example.com")}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + note(etag) |
| + if err := Release(ctx, f, &tw, Op{OfflineKindMove, "mv_a"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + _, cur, _ := f.current(OfflineActiveKey) |
| + note(cur) |
| + } |
| + |
| + g := newTableStore() |
| + old := plan3Body(t, map[string]OfflineReservation{"i-0a": moveEntry("mv_a", "a.example.com")}) |
| + g.set(OfflineActiveKey, old) |
| + if err := Release(ctx, g, &tw, Op{OfflineKindMove, "mv_a"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if body := readBody(t, g); len(body.Epoch) != 32 || body.Rev != 1 || len(body.Reservations) != 0 { |
| + t.Fatalf("a release of a plan 3 body wrote %+v", body) |
| + } |
| + // Two writes of one pre-epoch body get distinct epochs. |
| + h := newTableStore() |
| + h.set(OfflineActiveKey, old) |
| + if err := Release(ctx, h, &tw, Op{OfflineKindMove, "mv_a"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if readBody(t, h).Epoch == readBody(t, g).Epoch { |
| + t.Fatal("two fresh epochs are equal") |
| + } |
| +} |
| + |
| +func TestReleaseAlwaysWrites(t *testing.T) { |
| + ctx := context.Background() |
| + f := newTableStore() |
| + var tw TableWrites |
| + a, err := ReadActive(ctx, f, &tw) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + held := map[string]OfflineReservation{"i-0a": moveEntry("mv_a", "a.example.com")} |
| + if _, err := acquire(ctx, t, f, &tw, a, held); err != nil { |
| + t.Fatal(err) |
| + } |
| + before := readBody(t, f) |
| + if err := Release(ctx, f, &tw, Op{OfflineKindCleanup, "cu_nothing"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + after := readBody(t, f) |
| + if after.Rev != before.Rev+1 || after.Epoch != before.Epoch || len(after.Reservations) != 1 || after.Reservations["i-0a"] != held["i-0a"] { |
| + t.Fatalf("a release of an op holding nothing wrote %+v over %+v", after, before) |
| + } |
| + if !Holds(after.Reservations, []string{"i-0a"}, Op{OfflineKindMove, "mv_a"}) || Holds(after.Reservations, []string{"i-0a"}, Op{OfflineKindCleanup, "cu_nothing"}) || Holds(after.Reservations, []string{"i-0a", "i-0b"}, Op{OfflineKindMove, "mv_a"}) || Holds(after.Reservations, nil, Op{OfflineKindMove, "mv_a"}) { |
| + t.Fatal("Holds disagrees with the table") |
| + } |
| + // A malformed table is an error and is left as it is. |
| + bad := []byte(`{"v":1,"reservations":{"i-0a":{"kind":"lease"}}}`) |
| + f.set(OfflineActiveKey, bad) |
| + n := f.Requests() |
| + if err := Release(ctx, f, &tw, Op{OfflineKindMove, "mv_a"}); !errors.Is(err, ErrUnreadable) { |
| + t.Fatalf("a release over a malformed table = %v", err) |
| + } |
| + if b, _, _ := f.current(OfflineActiveKey); string(b) != string(bad) || f.Requests() != n { |
| + t.Fatal("a malformed table was written") |
| + } |
| +} |
| + |
| +// T4: an acquisition reads through ReadForAcquire, which normalises a |
| +// pre-epoch table first; only an epoch-bearing ETag is ever its condition. |
| +func TestAcquisitionNormalizesAPreEpochTableFirst(t *testing.T) { |
| + ctx := context.Background() |
| + f := newTableStore() |
| + var tw TableWrites |
| + entries := map[string]OfflineReservation{"i-0a": moveEntry("mv_op", "op.example.com")} |
| + e0 := f.set(OfflineActiveKey, plan3Body(t, entries)) |
| + a, err := ReadForAcquire(ctx, f, &tw) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if a.ETag == e0 || a.Body.Epoch == "" || a.Body.Rev != 1 || len(a.Table) != 1 || a.Table["i-0a"] != entries["i-0a"] { |
| + t.Fatalf("ReadForAcquire answered %+v at %s (pre-epoch %s)", a.Body, a.ETag, e0) |
| + } |
| + if _, err := acquire(ctx, t, f, &tw, a, map[string]OfflineReservation{HostKey("n.example.com"): {Kind: OfflineKindCreate, ID: "op_1", Host: "n.example.com", Account: "acct", App: "shop", ReservedAt: compatAt}}); err != nil { |
| + t.Fatal(err) |
| + } |
| + want := []string{"Get", "PutOnce(" + e0 + ")", "Get", "PutOnce(" + a.ETag + ")"} |
| + if got := f.Log(); strings.Join(got, " ") != strings.Join(want, " ") { |
| + t.Fatalf("log = %q, want %q", got, want) |
| + } |
| + norm := f.Sent()[0] |
| + var nb OfflineActive |
| + if err := json.Unmarshal(norm.body, &nb); err != nil || nb.Rev != 1 || nb.Epoch == "" || len(nb.Reservations) != 1 || nb.Reservations["i-0a"] != entries["i-0a"] { |
| + t.Fatalf("the normalisation wrote %s", norm.body) |
| + } |
| + |
| + // A normalisation that loses its condition is followed by a fresh Get, |
| + // never by an acquisition against the pre-epoch ETag. |
| + g := newTableStore() |
| + g0 := g.set(OfflineActiveKey, plan3Body(t, entries)) |
| + other := map[string]OfflineReservation{"i-0a": moveEntry("mv_op", "op.example.com"), "i-0b": moveEntry("mv_op2", "op2.example.com")} |
| + g.onPutOnce = func(n int, _ string, _ []byte, _ string) (bool, error) { |
| + if n == 0 { // a plan 3 writer lands first |
| + g.set(OfflineActiveKey, plan3Body(t, other)) |
| + } |
| + return true, nil |
| + } |
| + b, err := ReadForAcquire(ctx, g, &tw) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(b.Table) != 2 || b.Body.Epoch == "" || b.Body.Rev != 1 { |
| + t.Fatalf("after a lost normalisation ReadForAcquire answered %+v", b.Body) |
| + } |
| + log := g.Log() |
| + if len(log) != 5 || log[0] != "Get" || log[1] != "PutOnce("+g0+")" || log[2] != "Get" || log[4] != "Get" || log[3] == "PutOnce("+g0+")" { |
| + t.Fatalf("log = %q", log) |
| + } |
| + for _, p := range g.Sent() { |
| + var x OfflineActive |
| + _ = json.Unmarshal(p.body, &x) |
| + if len(x.Reservations) > len(other) { |
| + t.Fatal("an acquisition was sent") |
| + } |
| + } |
| + |
| + // A table that never takes the normalisation is unreadable after three |
| + // rounds. |
| + h := newTableStore() |
| + h.set(OfflineActiveKey, plan3Body(t, entries)) |
| + h.onPutOnce = func(int, string, []byte, string) (bool, error) { |
| + h.set(OfflineActiveKey, plan3Body(t, entries)) // and touched again |
| + return false, release.ErrCASConflict |
| + } |
| + if _, err := ReadForAcquire(ctx, h, &tw); !errors.Is(err, ErrUnreadable) || !strings.Contains(err.Error(), "could not be normalised") { |
| + t.Fatalf("ReadForAcquire over an unnormalisable table = %v", err) |
| + } |
| + if n := h.Requests(); n != 3 { |
| + t.Fatalf("%d normalisation writes, want 3", n) |
| + } |
| +} |
| + |
| +// V3: PeekActive only reads. |
| +func TestPeekActiveNeverWrites(t *testing.T) { |
| + ctx := context.Background() |
| + f := newTableStore() |
| + a, err := PeekActive(ctx, f) |
| + if err != nil || a.ETag != "" || a.Table == nil || len(a.Table) != 0 { |
| + t.Fatalf("PeekActive of an absent table = %+v %v", a, err) |
| + } |
| + entries := map[string]OfflineReservation{"i-0a": moveEntry("mv_a", "a.example.com")} |
| + e0 := f.set(OfflineActiveKey, plan3Body(t, entries)) |
| + a, err = PeekActive(ctx, f) |
| + if err != nil || a.ETag != e0 || a.Body.Epoch != "" || a.Body.Rev != 0 || a.Table["i-0a"] != entries["i-0a"] { |
| + t.Fatalf("PeekActive of a metadata-free body = %+v %v", a, err) |
| + } |
| + f.set(OfflineActiveKey, []byte(`{"v":1,"reservations":{"host/b.example.com":{"kind":"place","id":"pl","host":"a.example.com","account":"x","app":"y","reserved_at":"2026-10-24T09:00:00Z"}}}`)) |
| + if _, err := PeekActive(ctx, f); !errors.Is(err, ErrUnreadable) || !errors.Is(err, ErrOfflineActiveMalformed) { |
| + t.Fatalf("PeekActive of a malformed body = %v", err) |
| + } |
| + for _, l := range f.Log() { |
| + if l != "Get" { |
| + t.Fatalf("PeekActive sent %s", l) |
| + } |
| + } |
| +} |
| + |
| +// A store without PutOnce (the store door's client) is sent no table |
| +// write: ReadForAcquire and Release answer ErrNoSingleAttemptPut. |
| +func TestStoreWithoutPutOnceWritesNothing(t *testing.T) { |
| + var mu sync.Mutex |
| + var body []byte |
| + var methods []string |
| + door := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| + mu.Lock() |
| + defer mu.Unlock() |
| + methods = append(methods, r.Method) |
| + if r.Method != http.MethodGet || r.URL.Path != "/store/o/"+OfflineActiveKey { |
| + http.Error(w, "no", http.StatusMethodNotAllowed) |
| + return |
| + } |
| + if body == nil { |
| + http.NotFound(w, r) |
| + return |
| + } |
| + w.Header().Set("ETag", etagFor(body)) |
| + _, _ = w.Write(body) |
| + })) |
| + defer door.Close() |
| + s, err := fleetapi.New(door.URL, "fleet/box", "token") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + ctx := context.Background() |
| + var tw TableWrites |
| + for _, b := range [][]byte{nil, plan3Body(t, map[string]OfflineReservation{"i-0a": moveEntry("mv_a", "a.example.com")}), tableBody(t, OfflineActive{V: 1, Epoch: "00112233445566778899aabbccddeeff", Rev: 4})} { |
| + mu.Lock() |
| + body = b |
| + mu.Unlock() |
| + if _, err := ReadForAcquire(ctx, s, &tw); b != nil && strings.Contains(string(b), "epoch") { |
| + if err != nil { |
| + t.Fatalf("ReadForAcquire of a normalised table through the door = %v", err) |
| + } |
| + } else if !errors.Is(err, release.ErrNoSingleAttemptPut) { |
| + t.Fatalf("ReadForAcquire through the door = %v", err) |
| + } |
| + if err := Release(ctx, s, &tw, Op{OfflineKindMove, "mv_a"}); !errors.Is(err, release.ErrNoSingleAttemptPut) { |
| + t.Fatalf("Release through the door = %v", err) |
| + } |
| + } |
| + mu.Lock() |
| + defer mu.Unlock() |
| + for _, m := range methods { |
| + if m != http.MethodGet { |
| + t.Fatalf("the door saw %s: %q", m, methods) |
| + } |
| + } |
| + if tw.Unanswered() != 0 { |
| + t.Fatalf("%d writes left registered", tw.Unanswered()) |
| + } |
| +} |
diff --git a/internal/moves/tablewrite.go b/internal/moves/tablewrite.go| new file mode 100644 |
| index 000000000..de7d33415 |
| --- /dev/null |
| +++ b/internal/moves/tablewrite.go |
| @@ -0,0 +1,172 @@ |
| +package moves |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "sync" |
| + "sync/atomic" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/offlinecleanup" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// Single attempts, resent until answered (plan 4, Global Constraint "The |
| +// table is never deleted", F2). Every write of OfflineActiveKey is one |
| +// PutTable (TestEveryTableWriteGoesThroughPutTable). It sends one HTTP |
| +// request at a time through release.PutOnce, so no copy the code did not see |
| +// is ever in the network. An answer is returned: success, or |
| +// release.ErrCASConflict (our attempt landed or another write did; the |
| +// caller's next read tells which). A timeout, a transport error or a 5xx is |
| +// no answer, and the identical request (the same bytes, the same condition) |
| +// is sent again on offlinecleanup.RetrySchedule until it is answered or the |
| +// context ends. A write the context gave up on stays registered in its |
| +// process's TableWrites, and the process's Resend (the console's sweep and |
| +// drain, the armed box's run) keeps sending it until it is answered. |
| + |
| +// ErrUncertain is a table write whose context ended with no answer: it may |
| +// yet land. The write stays registered in its TableWrites and is resent |
| +// until answered; the caller keeps its state as if the write might have |
| +// landed, and the next successful conditional write by anyone settles it. |
| +var ErrUncertain = errors.New("moves: the reservation table write was not answered") |
| + |
| +// TableWrites is a process's table writes not yet answered: each with its |
| +// bytes, its condition and one owner. One mutex guards every registration; |
| +// the zero value is usable. |
| +// |
| +// The registration rule (review 8, item 6): a registration is removed only |
| +// by the goroutine that owns it, after that goroutine's final answer, under |
| +// the mutex. Ownership changes hands only under the mutex, and only while no |
| +// attempt of the write is inside PutOnce: PutTable owns its write while it |
| +// runs and gives it up (unowned) when its context ends; Resend claims only |
| +// unowned writes and gives back the ones it could not settle. So nothing |
| +// removes a registration while one of its attempts is still executing, and |
| +// Unanswered never reads zero while one is. |
| +type TableWrites struct { |
| + mu sync.Mutex |
| + writes []*tableWrite // registration order |
| + owners uint64 // the last owner token issued; 0 is unowned |
| +} |
| + |
| +type tableWrite struct { |
| + body []byte |
| + ifMatch string |
| + owner uint64 // 0: unowned |
| +} |
| + |
| +// Unanswered is how many writes are registered, whoever owns them. |
| +func (tw *TableWrites) Unanswered() int { |
| + tw.mu.Lock() |
| + defer tw.mu.Unlock() |
| + return len(tw.writes) |
| +} |
| + |
| +// register adds w owned by a new owner, under the mutex. |
| +func (tw *TableWrites) register(body []byte, ifMatch string) *tableWrite { |
| + tw.mu.Lock() |
| + defer tw.mu.Unlock() |
| + tw.owners++ |
| + w := &tableWrite{body: append([]byte(nil), body...), ifMatch: ifMatch, owner: tw.owners} |
| + tw.writes = append(tw.writes, w) |
| + return w |
| +} |
| + |
| +// settle removes w on an answer, or leaves it unowned, under the mutex. Only |
| +// w's owner calls it, after its last attempt has returned from PutOnce. |
| +func (tw *TableWrites) settle(w *tableWrite, answered bool) { |
| + tw.mu.Lock() |
| + defer tw.mu.Unlock() |
| + if !answered { |
| + w.owner = 0 |
| + return |
| + } |
| + for i, r := range tw.writes { |
| + if r == w { |
| + tw.writes = append(tw.writes[:i], tw.writes[i+1:]...) |
| + return |
| + } |
| + } |
| +} |
| + |
| +// Resend claims every unowned write under the mutex, sends each claimed |
| +// write once outside it (identically: the registered bytes and condition), |
| +// and under the mutex again removes each one that was answered and leaves |
| +// the others unowned. A write owned by a running PutTable or by another |
| +// Resend is skipped. |
| +func (tw *TableWrites) Resend(ctx context.Context, s release.Store) { |
| + tw.mu.Lock() |
| + tw.owners++ |
| + me := tw.owners |
| + var claimed []*tableWrite |
| + for _, w := range tw.writes { |
| + if w.owner == 0 { |
| + w.owner = me |
| + claimed = append(claimed, w) |
| + } |
| + } |
| + tw.mu.Unlock() |
| + for _, w := range claimed { |
| + _, err := sendTableWrite(ctx, s, w) |
| + tw.settle(w, tableAnswered(err)) |
| + } |
| +} |
| + |
| +// PutTable writes body to OfflineActiveKey under ifMatch ("" is create-only, |
| +// as release.Store.Put) and answers the new ETag, release.ErrCASConflict, |
| +// release.ErrNoSingleAttemptPut (nothing was sent: s has no PutOnce), or, |
| +// when ctx ends before any answer, ErrUncertain wrapped, with the write left |
| +// registered in tw, unowned, for tw.Resend. |
| +func PutTable(ctx context.Context, s release.Store, tw *TableWrites, body []byte, ifMatch string) (string, error) { |
| + w := tw.register(body, ifMatch) |
| + for n := 0; ; n++ { |
| + etag, err := sendTableWrite(ctx, s, w) |
| + if tableAnswered(err) { |
| + tw.settle(w, true) |
| + return etag, err |
| + } |
| + if ctx.Err() == nil { |
| + t := time.NewTimer(tableRetryWait(n)) |
| + select { |
| + case <-t.C: |
| + continue |
| + case <-ctx.Done(): |
| + t.Stop() |
| + } |
| + } |
| + tw.settle(w, false) |
| + return "", fmt.Errorf("%w: %w", ErrUncertain, err) |
| + } |
| +} |
| + |
| +// sendTableWrite is the table's one send: a single attempt of w, which |
| +// PutTable and Resend share. |
| +func sendTableWrite(ctx context.Context, s release.Store, w *tableWrite) (string, error) { |
| + return release.PutOnce(ctx, s, OfflineActiveKey, w.body, w.ifMatch) |
| +} |
| + |
| +// tableAnswered: success, a lost condition, or a store that sent nothing. |
| +func tableAnswered(err error) bool { |
| + return err == nil || errors.Is(err, release.ErrCASConflict) || errors.Is(err, release.ErrNoSingleAttemptPut) |
| +} |
| + |
| +// tableRetrySchedule is offlinecleanup.RetrySchedule unless a test set its |
| +// own (SetTableRetryScheduleForTest). |
| +var tableRetrySchedule atomic.Pointer[offlinecleanup.Backoff] |
| + |
| +func tableRetryWait(n int) time.Duration { |
| + if b := tableRetrySchedule.Load(); b != nil { |
| + return b.Wait(n) |
| + } |
| + return offlinecleanup.RetrySchedule.Wait(n) |
| +} |
| + |
| +// SetTableRetryScheduleForTest makes PutTable resend after steps, then |
| +// every every, instead of 1 s, 2 s, 4 s and every 10 s, until the returned |
| +// restore runs. A test seam only (TestEveryE2ESeamHasOneSetter lists its |
| +// setters). |
| +func SetTableRetryScheduleForTest(every time.Duration, steps ...time.Duration) (restore func()) { |
| + b := offlinecleanup.NewBackoff(every, steps...) |
| + prev := tableRetrySchedule.Swap(&b) |
| + return func() { tableRetrySchedule.Store(prev) } |
| +} |
diff --git a/internal/moves/tablewrite_audit_test.go b/internal/moves/tablewrite_audit_test.go| new file mode 100644 |
| index 000000000..43556d1d6 |
| --- /dev/null |
| +++ b/internal/moves/tablewrite_audit_test.go |
| @@ -0,0 +1,211 @@ |
| +package moves |
| + |
| +import ( |
| + "fmt" |
| + "go/ast" |
| + "go/constant" |
| + "go/types" |
| + "os" |
| + "path/filepath" |
| + "sort" |
| + "strings" |
| + "testing" |
| + |
| + "golang.org/x/tools/go/packages" |
| +) |
| + |
| +// tableWriteSite is one call that writes or deletes OfflineActiveKey: the |
| +// function it sits in and what it calls, both by resolved symbol with the |
| +// module's internal/ prefix trimmed ("(*console.Server).reserveBoxes", |
| +// "release.PutOnce", "method Put"). |
| +type tableWriteSite struct{ Func, Call string } |
| + |
| +// tableWritePermitted is every call allowed to write the table, with how |
| +// many times it may appear. The send inside PutTable (sendTableWrite, which |
| +// PutTable and TableWrites.Resend share) is the one permanent entry. The |
| +// console's three sites are plan 2's reserveBoxes and releaseBoxes, kept by |
| +// an explicit list until Task 2b rewrites them over PutTable and empties |
| +// it (ruling M3); an entry whose call no longer exists fails the audit, so |
| +// the list cannot outlive them. |
| +var tableWritePermitted = map[tableWriteSite]int{ |
| + {"moves.sendTableWrite", "release.PutOnce"}: 1, |
| + // Task 2b removes these three (offlinereserve.go:241, :478, :487). |
| + {"(*console.Server).reserveBoxes", "method Put"}: 1, |
| + {"(*console.Server).releaseBoxes", "release.DeleteIfMatch"}: 1, |
| + {"(*console.Server).releaseBoxes", "method Put"}: 1, |
| +} |
| + |
| +// The calls that write or delete a key: the Store methods and the release |
| +// helpers that take one. |
| +var tableWriteCalls = map[string]bool{"Put": true, "PutFrom": true, "PutOnce": true, "Delete": true, "DeleteIfMatch": true, "PutObjectFrom": true} |
| + |
| +const auditModule = "github.com/carlosframework/platform/internal/" |
| + |
| +type tableWriteFinding struct { |
| + site tableWriteSite |
| + file string |
| + line int |
| +} |
| + |
| +// findTableWrites walks pkgs' syntax for a call named in tableWriteCalls |
| +// any of whose arguments is a constant string equal to OfflineActiveKey |
| +// (moves.OfflineActiveKey, any other constant of that value, or the |
| +// literal). |
| +func findTableWrites(pkgs []*packages.Package) []tableWriteFinding { |
| + var out []tableWriteFinding |
| + trim := func(s string) string { return strings.ReplaceAll(s, auditModule, "") } |
| + for _, p := range pkgs { |
| + for _, f := range p.Syntax { |
| + for _, decl := range f.Decls { |
| + fn := "<package scope>" |
| + if d, ok := decl.(*ast.FuncDecl); ok { |
| + if obj, ok := p.TypesInfo.Defs[d.Name].(*types.Func); ok { |
| + fn = trim(obj.FullName()) |
| + } |
| + } |
| + ast.Inspect(decl, func(n ast.Node) bool { |
| + call, ok := n.(*ast.CallExpr) |
| + if !ok { |
| + return true |
| + } |
| + var id *ast.Ident |
| + switch fun := call.Fun.(type) { |
| + case *ast.SelectorExpr: |
| + id = fun.Sel |
| + case *ast.Ident: |
| + id = fun |
| + default: |
| + return true |
| + } |
| + if !tableWriteCalls[id.Name] { |
| + return true |
| + } |
| + keyed := false |
| + for _, a := range call.Args { |
| + tv, ok := p.TypesInfo.Types[a] |
| + if ok && tv.Value != nil && tv.Value.Kind() == constant.String && constant.StringVal(tv.Value) == OfflineActiveKey { |
| + keyed = true |
| + } |
| + } |
| + if !keyed { |
| + return true |
| + } |
| + what := "method " + id.Name |
| + if obj, ok := p.TypesInfo.Uses[id].(*types.Func); ok { |
| + if sig, ok := obj.Type().(*types.Signature); ok && sig.Recv() == nil && obj.Pkg() != nil { |
| + what = trim(obj.Pkg().Path()) + "." + obj.Name() |
| + } |
| + } |
| + pos := p.Fset.Position(call.Pos()) |
| + out = append(out, tableWriteFinding{tableWriteSite{fn, what}, pos.Filename, pos.Line}) |
| + return true |
| + }) |
| + } |
| + } |
| + } |
| + return out |
| +} |
| + |
| +// Every write of control/offline-moves/active.json goes through PutTable |
| +// (Global Constraint "The table is never deleted", F2): no Put, PutFrom, |
| +// PutOnce, Delete or DeleteIfMatch names the key in the non-test files |
| +// under internal/ outside PutTable's one send, and the console's plan 2 |
| +// sites until Task 2b. A fixture overlaid beside the real files holds one |
| +// copy of each forbidden shape, which must all be reported. |
| +func TestEveryTableWriteGoesThroughPutTable(t *testing.T) { |
| + root, err := filepath.Abs(filepath.Join("..", "..")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + fixture := filepath.Join(root, "internal", "moves", "zz_tablewrite_fixture.go") |
| + cfg := &packages.Config{ |
| + Dir: root, |
| + Mode: packages.NeedName | packages.NeedFiles | packages.NeedCompiledGoFiles | packages.NeedSyntax | packages.NeedTypes | packages.NeedTypesInfo | packages.NeedImports, |
| + Env: append(os.Environ(), "GOOS=linux", "GOARCH=amd64", "CGO_ENABLED=0"), |
| + Overlay: map[string][]byte{fixture: []byte(tableWriteFixture)}, |
| + } |
| + pkgs, err := packages.Load(cfg, "./internal/...") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if packages.PrintErrors(pkgs) > 0 { |
| + t.Fatal("loading internal/... failed") |
| + } |
| + found := findTableWrites(pkgs) |
| + if len(found) == 0 { |
| + t.Fatal("the audit found no write of the table at all; it has stopped reading the code") |
| + } |
| + counts := map[tableWriteSite]int{} |
| + var bad, fixed []string |
| + for _, f := range found { |
| + if f.file == fixture { |
| + fixed = append(fixed, f.site.Func+" -> "+f.site.Call) |
| + continue |
| + } |
| + rel, _ := filepath.Rel(root, f.file) |
| + if _, ok := tableWritePermitted[f.site]; ok { |
| + counts[f.site]++ |
| + continue |
| + } |
| + bad = append(bad, fmt.Sprintf("%s:%d: %s -> %s", filepath.ToSlash(rel), f.line, f.site.Func, f.site.Call)) |
| + } |
| + if len(bad) > 0 { |
| + t.Errorf("writes of %s outside moves.PutTable (write through PutTable):\n%s", OfflineActiveKey, strings.Join(bad, "\n")) |
| + } |
| + for site, want := range tableWritePermitted { |
| + if counts[site] != want { |
| + t.Errorf("%s -> %s appears %d times, its entry permits %d (remove an entry whose call is gone)", site.Func, site.Call, counts[site], want) |
| + } |
| + } |
| + want := []string{ |
| + "moves.auditFixtureDelete -> method Delete", |
| + "moves.auditFixtureDeleteIfMatch -> release.DeleteIfMatch", |
| + "moves.auditFixtureLiteral -> method Put", |
| + "moves.auditFixtureOtherConstant -> method PutFrom", |
| + "moves.auditFixturePutOnce -> release.PutOnce", |
| + "moves.auditFixtureStreamed -> release.PutObjectFrom", |
| + } |
| + sort.Strings(fixed) |
| + if strings.Join(fixed, "\n") != strings.Join(want, "\n") { |
| + t.Errorf("the fixture's findings:\n%s\nwant:\n%s", strings.Join(fixed, "\n"), strings.Join(want, "\n")) |
| + } |
| +} |
| + |
| +const tableWriteFixture = `package moves |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +const auditFixtureKey = "control/offline-moves/" + "active.json" |
| + |
| +func auditFixtureLiteral(ctx context.Context, s release.Store) { |
| + _, _ = s.Put(ctx, "control/offline-moves/active.json", nil, "") |
| +} |
| + |
| +func auditFixtureDelete(ctx context.Context, s release.Store) { _ = s.Delete(ctx, OfflineActiveKey) } |
| + |
| +func auditFixtureDeleteIfMatch(ctx context.Context, s release.Store) { |
| + _ = release.DeleteIfMatch(ctx, s, OfflineActiveKey, "\"e\"") |
| +} |
| + |
| +func auditFixturePutOnce(ctx context.Context, s release.Store) { |
| + _, _ = release.PutOnce(ctx, s, OfflineActiveKey, nil, "") |
| +} |
| + |
| +func auditFixtureOtherConstant(ctx context.Context, s release.PutStreamer) { |
| + _, _ = s.PutFrom(ctx, auditFixtureKey, bytes.NewReader(nil), 0, "") |
| +} |
| + |
| +func auditFixtureStreamed(ctx context.Context, s release.Store) { |
| + _, _ = release.PutObjectFrom(ctx, s, OfflineActiveKey, bytes.NewReader(nil), 0, "") |
| +} |
| + |
| +// Another key is not the table. |
| +func auditFixtureOtherKey(ctx context.Context, s release.Store) { |
| + _, _ = s.Put(ctx, "control/offline-moves/other.json", nil, "") |
| +} |
| +` |
diff --git a/internal/moves/tablewrite_test.go b/internal/moves/tablewrite_test.go| new file mode 100644 |
| index 000000000..b631768cb |
| --- /dev/null |
| +++ b/internal/moves/tablewrite_test.go |
| @@ -0,0 +1,171 @@ |
| +package moves |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + "errors" |
| + "sync" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// An attempt with no answer is sent again identically; a 412 after it is an |
| +// answer; a context that ends first leaves the write registered, and a |
| +// later Resend that is answered settles it. |
| +func TestUnansweredTableWriteIsResentIdentically(t *testing.T) { |
| + fastResend(t) |
| + ctx := context.Background() |
| + f := newTableStore() |
| + etag := f.set(OfflineActiveKey, tableBody(t, OfflineActive{V: 1, Epoch: "00112233445566778899aabbccddeeff", Rev: 1})) |
| + f.onPutOnce = func(n int, _ string, _ []byte, _ string) (bool, error) { |
| + if n == 0 { |
| + return false, errTimeout |
| + } |
| + f.set(OfflineActiveKey, tableBody(t, OfflineActive{V: 1, Epoch: "00112233445566778899aabbccddeeff", Rev: 2})) // another writer landed |
| + return true, nil |
| + } |
| + var tw TableWrites |
| + body := tableBody(t, OfflineActive{V: 1, Epoch: "00112233445566778899aabbccddeeff", Rev: 2, Reservations: map[string]OfflineReservation{"i-0a": moveEntry("mv_a", "a.example.com")}}) |
| + if _, err := PutTable(ctx, f, &tw, body, etag); !errors.Is(err, release.ErrCASConflict) { |
| + t.Fatalf("a 412 after an unanswered attempt = %v, want ErrCASConflict", err) |
| + } |
| + sent := f.Sent() |
| + if len(sent) != 2 || !bytes.Equal(sent[0].body, sent[1].body) || !bytes.Equal(sent[0].body, body) || sent[0].ifMatch != etag || sent[1].ifMatch != etag || sent[0].key != OfflineActiveKey || sent[1].key != OfflineActiveKey { |
| + t.Fatalf("the resend was not identical: %+v", sent) |
| + } |
| + if tw.Unanswered() != 0 { |
| + t.Fatalf("%d writes registered after an answer", tw.Unanswered()) |
| + } |
| + |
| + // No answer before the context ends: uncertain, still registered. |
| + g := newTableStore() |
| + e1 := g.set(OfflineActiveKey, tableBody(t, OfflineActive{V: 1, Epoch: "00112233445566778899aabbccddeeff", Rev: 1})) |
| + var mu sync.Mutex |
| + answering := false |
| + g.onPutOnce = func(int, string, []byte, string) (bool, error) { |
| + mu.Lock() |
| + defer mu.Unlock() |
| + if answering { |
| + return true, nil |
| + } |
| + return false, errTimeout |
| + } |
| + short, cancel := context.WithTimeout(ctx, 30*time.Millisecond) |
| + defer cancel() |
| + if _, err := PutTable(short, g, &tw, body, e1); !errors.Is(err, ErrUncertain) { |
| + t.Fatalf("an unanswered write = %v, want ErrUncertain", err) |
| + } |
| + if tw.Unanswered() != 1 { |
| + t.Fatalf("Unanswered = %d, want 1", tw.Unanswered()) |
| + } |
| + n := g.Requests() |
| + if n < 2 { |
| + t.Fatalf("the write was sent %d times before its context ended", n) |
| + } |
| + // A Resend that is not answered leaves it registered and unowned. |
| + tw.Resend(ctx, g) |
| + if tw.Unanswered() != 1 || g.Requests() != n+1 { |
| + t.Fatalf("an unanswered Resend: Unanswered %d, requests %d", tw.Unanswered(), g.Requests()) |
| + } |
| + mu.Lock() |
| + answering = true |
| + mu.Unlock() |
| + tw.Resend(ctx, g) |
| + if tw.Unanswered() != 0 { |
| + t.Fatalf("Unanswered = %d after an answered Resend", tw.Unanswered()) |
| + } |
| + last := g.Sent()[g.Requests()-1] |
| + if !bytes.Equal(last.body, body) || last.ifMatch != e1 { |
| + t.Fatal("the Resend was not the registered write") |
| + } |
| + if b, _, _ := g.current(OfflineActiveKey); !bytes.Equal(b, body) { |
| + t.Fatal("the answered Resend did not land") |
| + } |
| + // Nothing registered: Resend sends nothing. |
| + tw.Resend(ctx, g) |
| + if g.Requests() != n+2 { |
| + t.Fatal("an empty Resend sent a request") |
| + } |
| +} |
| + |
| +// Review 8, item 6: a registration is removed only by its owner, after its |
| +// final answer, and ownership changes hands only while no attempt of the |
| +// write is inside PutOnce. Run under -race. |
| +func TestRegistrationIsRemovedOnlyByItsOwner(t *testing.T) { |
| + fastResend(t) |
| + ctx := context.Background() |
| + f := newTableStore() |
| + e1 := f.set(OfflineActiveKey, tableBody(t, OfflineActive{V: 1, Epoch: "00112233445566778899aabbccddeeff", Rev: 1})) |
| + type held struct { |
| + n int |
| + release chan error |
| + } |
| + inside := make(chan held) |
| + f.onPutOnce = func(n int, _ string, _ []byte, _ string) (bool, error) { |
| + h := held{n, make(chan error)} |
| + inside <- h |
| + return false, <-h.release // the answer the test hands it |
| + } |
| + var tw TableWrites |
| + body := tableBody(t, OfflineActive{V: 1, Epoch: "00112233445566778899aabbccddeeff", Rev: 2, Reservations: map[string]OfflineReservation{"i-0a": moveEntry("mv_a", "a.example.com")}}) |
| + callerCtx, endCaller := context.WithCancel(ctx) |
| + done := make(chan error, 1) |
| + go func() { |
| + _, err := PutTable(callerCtx, f, &tw, body, e1) |
| + done <- err |
| + }() |
| + first := <-inside // the caller's first attempt is inside PutOnce |
| + for range 2 { |
| + tw.Resend(ctx, f) // owned by the caller: skipped, nothing sent |
| + if f.Requests() != 1 || tw.Unanswered() != 1 { |
| + t.Fatalf("while the caller's attempt is held: requests %d, Unanswered %d", f.Requests(), tw.Unanswered()) |
| + } |
| + } |
| + endCaller() |
| + first.release <- errTimeout |
| + if err := <-done; !errors.Is(err, ErrUncertain) { |
| + t.Fatalf("the caller = %v, want ErrUncertain", err) |
| + } |
| + if f.Requests() != 1 || tw.Unanswered() != 1 { |
| + t.Fatalf("after the caller gave up: requests %d, Unanswered %d", f.Requests(), tw.Unanswered()) |
| + } |
| + |
| + // A Resend claims it; while its attempt is held a second Resend sends |
| + // nothing. |
| + resent := make(chan struct{}) |
| + go func() { tw.Resend(ctx, f); close(resent) }() |
| + second := <-inside |
| + if second.n != 1 { |
| + t.Fatalf("the held attempt is request %d", second.n) |
| + } |
| + tw.Resend(ctx, f) |
| + if f.Requests() != 2 || tw.Unanswered() != 1 { |
| + t.Fatalf("while a Resend's attempt is held: requests %d, Unanswered %d", f.Requests(), tw.Unanswered()) |
| + } |
| + second.release <- release.ErrCASConflict |
| + <-resent |
| + if tw.Unanswered() != 0 { |
| + t.Fatalf("Unanswered = %d after the held Resend's 412", tw.Unanswered()) |
| + } |
| +} |
| + |
| +// The resend schedule is offlinecleanup.RetrySchedule unless a test sets |
| +// its own. |
| +func TestTableResendScheduleIsTheSharedOne(t *testing.T) { |
| + if got := tableRetryWait(0); got != time.Second { |
| + t.Fatalf("the first resend waits %v", got) |
| + } |
| + if got := tableRetryWait(3); got != 10*time.Second { |
| + t.Fatalf("the fourth resend waits %v", got) |
| + } |
| + restore := SetTableRetryScheduleForTest(time.Millisecond, 2*time.Millisecond) |
| + if tableRetryWait(0) != 2*time.Millisecond || tableRetryWait(5) != time.Millisecond { |
| + t.Fatal("the test schedule is not used") |
| + } |
| + restore() |
| + if tableRetryWait(1) != 2*time.Second { |
| + t.Fatal("restore did not put the shared schedule back") |
| + } |
| +} |
diff --git a/internal/release/putonce_test.go b/internal/release/putonce_test.go| new file mode 100644 |
| index 000000000..8c7307fef |
| --- /dev/null |
| +++ b/internal/release/putonce_test.go |
| @@ -0,0 +1,174 @@ |
| +package release |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "net/http" |
| + "net/http/httptest" |
| + "sync" |
| + "testing" |
| + |
| + "github.com/aws/aws-sdk-go-v2/aws" |
| + "github.com/aws/aws-sdk-go-v2/credentials" |
| + awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| +) |
| + |
| +// putCounter is an S3 endpoint that answers every PutObject with status |
| +// and counts the requests that reached it, with each one's If-Match and |
| +// If-None-Match. |
| +type putCounter struct { |
| + mu sync.Mutex |
| + status int |
| + code string |
| + requests []http.Header |
| +} |
| + |
| +func (p *putCounter) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| + if r.Method != http.MethodPut { |
| + http.Error(w, "unexpected", http.StatusMethodNotAllowed) |
| + return |
| + } |
| + _, _ = io.Copy(io.Discard, r.Body) |
| + p.mu.Lock() |
| + p.requests = append(p.requests, r.Header.Clone()) |
| + status, code := p.status, p.code |
| + p.mu.Unlock() |
| + if status == http.StatusOK { |
| + w.Header().Set("ETag", `"landed"`) |
| + w.WriteHeader(http.StatusOK) |
| + return |
| + } |
| + w.Header().Set("Content-Type", "application/xml") |
| + w.WriteHeader(status) |
| + fmt.Fprintf(w, `<Error><Code>%s</Code><Message>fixture</Message></Error>`, code) |
| +} |
| + |
| +func (p *putCounter) count() int { |
| + p.mu.Lock() |
| + defer p.mu.Unlock() |
| + return len(p.requests) |
| +} |
| + |
| +// defaultRetryS3 is an S3Store whose client keeps the SDK's default |
| +// retryer, as newS3Store's client does: nothing here sets a retry option. |
| +func defaultRetryS3(url string) *S3Store { |
| + return &S3Store{Bucket: "bucket", Client: awss3.New(awss3.Options{ |
| + Region: "us-east-1", |
| + Credentials: credentials.NewStaticCredentialsProvider("fixture", "fixture", ""), |
| + BaseEndpoint: aws.String(url), |
| + UsePathStyle: true, |
| + })} |
| +} |
| + |
| +// A table write is one HTTP attempt: PutOnce returns a 5xx after exactly |
| +// one request reached the server, where Put sends the SDK's default three; |
| +// a 412 is release.ErrCASConflict after one request. |
| +func TestTableWriteIsOneHTTPAttempt(t *testing.T) { |
| + ctx := context.Background() |
| + srv500 := &putCounter{status: http.StatusInternalServerError, code: "InternalError"} |
| + h := httptest.NewServer(srv500) |
| + defer h.Close() |
| + s := defaultRetryS3(h.URL) |
| + |
| + if _, err := s.PutOnce(ctx, "control/key.json", []byte("{}"), `"e1"`); err == nil || errors.Is(err, ErrCASConflict) { |
| + t.Fatalf("PutOnce against a 500 = %v, want the server's error", err) |
| + } |
| + if n := srv500.count(); n != 1 { |
| + t.Fatalf("PutOnce sent %d requests, want exactly 1", n) |
| + } |
| + if got := srv500.requests[0].Get("If-Match"); got != `"e1"` { |
| + t.Fatalf("PutOnce's If-Match = %q", got) |
| + } |
| + // Put keeps the client's retryer: the SDK's default three attempts. |
| + if _, err := s.Put(ctx, "control/key.json", []byte("{}"), `"e1"`); err == nil { |
| + t.Fatal("Put against a 500 succeeded") |
| + } |
| + if n := srv500.count(); n != 1+3 { |
| + t.Fatalf("Put sent %d requests, want the default 3", n-1) |
| + } |
| + |
| + srv412 := &putCounter{status: http.StatusPreconditionFailed, code: "PreconditionFailed"} |
| + h412 := httptest.NewServer(srv412) |
| + defer h412.Close() |
| + s412 := defaultRetryS3(h412.URL) |
| + if _, err := s412.PutOnce(ctx, "control/key.json", []byte("{}"), ""); !errors.Is(err, ErrCASConflict) { |
| + t.Fatalf("PutOnce against a 412 = %v, want ErrCASConflict", err) |
| + } |
| + if n := srv412.count(); n != 1 { |
| + t.Fatalf("PutOnce against a 412 sent %d requests, want 1", n) |
| + } |
| + if got := srv412.requests[0].Get("If-None-Match"); got != "*" { |
| + t.Fatalf("a create-only PutOnce's If-None-Match = %q, want *", got) |
| + } |
| + // A replace-only write whose object is gone (404) is a CAS conflict, |
| + // mapped as Put maps it. |
| + srv404 := &putCounter{status: http.StatusNotFound, code: "NoSuchKey"} |
| + h404 := httptest.NewServer(srv404) |
| + defer h404.Close() |
| + if _, err := defaultRetryS3(h404.URL).PutOnce(ctx, "control/key.json", []byte("{}"), `"e1"`); !errors.Is(err, ErrCASConflict) { |
| + t.Fatalf("PutOnce against a 404 under If-Match = %v, want ErrCASConflict", err) |
| + } |
| + |
| + // Success answers the ETag. |
| + ok := &putCounter{status: http.StatusOK} |
| + hok := httptest.NewServer(ok) |
| + defer hok.Close() |
| + if etag, err := NewPrefixStore(defaultRetryS3(hok.URL), "scope").(SingleAttemptPutter).PutOnce(ctx, "k", []byte("{}"), ""); err != nil || etag != `"landed"` || ok.count() != 1 { |
| + t.Fatalf("PutOnce through a prefix store = %q %v after %d requests", etag, err, ok.count()) |
| + } |
| +} |
| + |
| +type onceLessStore struct{ Store } |
| + |
| +// PutOnce reaches the store's own single attempt, through any prefix |
| +// wrapper; a store without one is ErrNoSingleAttemptPut and is sent nothing. |
| +func TestPutOnceDispatch(t *testing.T) { |
| + ctx := context.Background() |
| + d := NewDirStore(t.TempDir()) |
| + etag, err := PutOnce(ctx, d, "a/key.json", []byte("one"), "") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := PutOnce(ctx, d, "a/key.json", []byte("two"), ""); !errors.Is(err, ErrCASConflict) { |
| + t.Fatalf("a create-only PutOnce over an object = %v", err) |
| + } |
| + if _, err := PutOnce(ctx, d, "a/key.json", []byte("two"), `"stale"`); !errors.Is(err, ErrCASConflict) { |
| + t.Fatalf("a stale PutOnce = %v", err) |
| + } |
| + if _, err := PutOnce(ctx, d, "a/key.json", []byte("two"), etag); err != nil { |
| + t.Fatal(err) |
| + } |
| + if b, _, _ := d.Get(ctx, "a/key.json"); string(b) != "two" { |
| + t.Fatalf("DirStore holds %q", b) |
| + } |
| + |
| + scoped := NewPrefixStore(NewPrefixStore(d, "outer"), "inner") |
| + if _, err := PutOnce(ctx, scoped, "k", []byte("scoped"), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + if b, _, err := d.Get(ctx, "outer/inner/k"); err != nil || string(b) != "scoped" { |
| + t.Fatalf("the prefix store wrote %q %v", b, err) |
| + } |
| + |
| + trap := &putCountingStore{Store: d} |
| + for _, s := range []Store{onceLessStore{trap}, NewPrefixStore(onceLessStore{trap}, "scope")} { |
| + if _, err := PutOnce(ctx, s, "k", []byte("x"), ""); !errors.Is(err, ErrNoSingleAttemptPut) { |
| + t.Fatalf("PutOnce on a store without it = %v", err) |
| + } |
| + } |
| + if trap.puts != 0 { |
| + t.Fatalf("a store without PutOnce was sent %d puts", trap.puts) |
| + } |
| +} |
| + |
| +type putCountingStore struct { |
| + Store |
| + puts int |
| +} |
| + |
| +func (c *putCountingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + c.puts++ |
| + return c.Store.Put(ctx, key, body, ifMatch) |
| +} |
diff --git a/internal/release/store.go b/internal/release/store.go| index 3a836d25a..5d1546134 100644 |
| --- a/internal/release/store.go |
| +++ b/internal/release/store.go |
| @@ -56,6 +56,37 @@ type Store interface { |
| Delete(ctx context.Context, key string) error |
| } |
| +// SingleAttemptPutter is the optional half of Store that sends a conditional |
| +// Put as exactly one request: no retry the caller did not see. A write whose |
| +// every copy must be accounted for (the offline moves' reservation table, |
| +// moves.PutTable) resends through this, so a copy can only ever be one the |
| +// caller sent and is still waiting on. |
| +type SingleAttemptPutter interface { |
| + PutOnce(ctx context.Context, key string, body []byte, ifMatch string) (string, error) |
| +} |
| + |
| +// ErrNoSingleAttemptPut is a store that cannot send a single attempt (the |
| +// store door's client, whose door writes through a retrying client): |
| +// nothing was sent. |
| +var ErrNoSingleAttemptPut = errors.New("store: no single-attempt put") |
| + |
| +// PutOnce is s's PutOnce, with Put's conditions and answers, or |
| +// ErrNoSingleAttemptPut with nothing sent when s has none. Put itself is |
| +// unchanged for every other caller. |
| +func PutOnce(ctx context.Context, s Store, key string, body []byte, ifMatch string) (string, error) { |
| + once, ok := s.(SingleAttemptPutter) |
| + if !ok { |
| + return "", ErrNoSingleAttemptPut |
| + } |
| + return once.PutOnce(ctx, key, body, ifMatch) |
| +} |
| + |
| +var ( |
| + _ SingleAttemptPutter = (*S3Store)(nil) |
| + _ SingleAttemptPutter = (*DirStore)(nil) |
| + _ SingleAttemptPutter = prefixStore{} |
| +) |
| + |
| // ---- prefixStore ------------------------------------------------------------- |
| // prefixStore rebases every key a caller works with onto a fixed base before |
| @@ -142,6 +173,12 @@ func (p prefixStore) PutFrom(ctx context.Context, key string, body io.ReadSeeker |
| return PutObjectFrom(ctx, p.Store, p.full(key), body, size, ifMatch) |
| } |
| +// PutOnce is written out, not promoted, for the reason Open is; a store |
| +// underneath without one answers ErrNoSingleAttemptPut. |
| +func (p prefixStore) PutOnce(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + return PutOnce(ctx, p.Store, p.full(key), body, ifMatch) |
| +} |
| + |
| func (p prefixStore) Delete(ctx context.Context, key string) error { |
| return p.Store.Delete(ctx, p.full(key)) |
| } |
| @@ -453,6 +490,11 @@ func (d *DirStore) Put(_ context.Context, key string, body []byte, ifMatch strin |
| // PutFrom is Put streamed: the body goes to a temp file (hashed as it |
| // goes, so the etag is the same content sha256 Put would give) and is |
| // renamed into place under d.mu after the same CAS gate. |
| +// PutOnce is Put: a local write has no retries. |
| +func (d *DirStore) PutOnce(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + return d.Put(ctx, key, body, ifMatch) |
| +} |
| + |
| func (d *DirStore) PutFrom(_ context.Context, key string, body io.ReadSeeker, size int64, ifMatch string) (string, error) { |
| p, err := d.path(key) |
| if err != nil { |
| @@ -785,6 +827,19 @@ func (s *S3Store) Open(ctx context.Context, key string) (io.ReadCloser, string, |
| } |
| func (s *S3Store) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + return s.put(ctx, key, body, ifMatch) |
| +} |
| + |
| +// PutOnce is Put sent as one HTTP attempt: the per-call NopRetryer replaces |
| +// the client's default retryer for this call only (MaxAttempts 1, nothing |
| +// retried), so a timeout, a transport error or a 5xx comes back to the |
| +// caller after the one request, never resent by the SDK. The condition and |
| +// the error mapping are Put's. |
| +func (s *S3Store) PutOnce(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + return s.put(ctx, key, body, ifMatch, func(o *awss3.Options) { o.Retryer = aws.NopRetryer{} }) |
| +} |
| + |
| +func (s *S3Store) put(ctx context.Context, key string, body []byte, ifMatch string, optFns ...func(*awss3.Options)) (string, error) { |
| in := &awss3.PutObjectInput{ |
| Bucket: &s.Bucket, |
| Key: &key, |
| @@ -796,7 +851,7 @@ func (s *S3Store) Put(ctx context.Context, key string, body []byte, ifMatch stri |
| } else { |
| in.IfMatch = aws.String(ifMatch) // replace-only: 412 if the object moved on |
| } |
| - out, err := s.Client.PutObject(ctx, in) |
| + out, err := s.Client.PutObject(ctx, in, optFns...) |
| if err != nil { |
| // A replace-only write whose object has been DELETED under it comes |
| // back 404/NoSuchKey, not 412: there is no etag left to fail the |