amadan
settings-4 task 1: the build state, its registry copy, and root's consumed file
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2 files changed,
+336
−0
internal/release/buildstate.go+184 −0internal/release/buildstate_test.go+152 −0
diff --git a/internal/release/buildstate.go b/internal/release/buildstate.go| new file mode 100644 |
| index 000000000..6a1fb7a7c |
| --- /dev/null |
| +++ b/internal/release/buildstate.go |
| @@ -0,0 +1,184 @@ |
| +package release |
| + |
| +import ( |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "os" |
| + "path/filepath" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/controlname" |
| +) |
| + |
| +// ResolvedRequest is an install request R resolved to a build T (settings |
| +// spec §4.3 rule 2). |
| +type ResolvedRequest struct { |
| + ID uint64 `json:"id"` |
| + Target string `json:"target"` |
| +} |
| + |
| +// BuildState is the object beside an instance's replica (settings spec §4.2): |
| +// the build whose process last opened the database, the request root last |
| +// resolved, and the highest request completed. The agent writes it under the |
| +// instance's lease before any process runs a build against the database; it |
| +// is trusted the way settings are (§2 decision 5). |
| +type BuildState struct { |
| + Schema int `json:"schema"` // 1 |
| + Host string `json:"host"` |
| + LastExecuted string `json:"last_executed,omitempty"` |
| + Resolved *ResolvedRequest `json:"resolved,omitempty"` |
| + Completed uint64 `json:"completed,omitempty"` |
| + CompletedFailed bool `json:"completed_failed,omitempty"` |
| + At time.Time `json:"at"` |
| +} |
| + |
| +// BuildStateKey is beside the lease's lock.json at the same prefix. |
| +func BuildStateKey(replicaPrefix string) string { return replicaPrefix + "/build-state.json" } |
| + |
| +// BuildStateCopy is the agent's registry copy, refreshed at every reconcile |
| +// and before every wake. Known is false until the agent has read the bucket |
| +// object (absent counts as read); Replica is whether the instance has a |
| +// replica, which decides rule 3's "unknown". |
| +type BuildStateCopy struct { |
| + Known bool `json:"known"` |
| + Replica bool `json:"replica,omitempty"` |
| + State *BuildState `json:"state,omitempty"` |
| + ReadAt time.Time `json:"read_at"` |
| + // ReadErr is set when the latest refresh could not read the object; the |
| + // copy keeps its last known state, and the box reports the host as |
| + // incomplete for retention (Astra I1 #6). Cleared by a good read. |
| + ReadErr string `json:"read_err,omitempty"` |
| +} |
| + |
| +func ParseBuildStateCopy(col string) (BuildStateCopy, error) { |
| + if col == "" { |
| + return BuildStateCopy{}, nil |
| + } |
| + var c BuildStateCopy |
| + if err := json.Unmarshal([]byte(col), &c); err != nil { |
| + return BuildStateCopy{}, fmt.Errorf("build state copy: %v", err) |
| + } |
| + return c, nil |
| +} |
| + |
| +func (c BuildStateCopy) Column() string { |
| + b, _ := json.Marshal(c) |
| + return string(b) |
| +} |
| + |
| +// Consumed is root's own record for one host at /opt/carlos/consumed/<host> |
| +// (settings spec §4.3 rule 2): the request it resolved, before staging, and |
| +// the highest it saw complete, plus its last decision for reports. Root-owned; |
| +// the agent reads it. |
| +type Consumed struct { |
| + Schema int `json:"schema"` // 1 |
| + Host string `json:"host"` |
| + Resolved *ResolvedRequest `json:"resolved,omitempty"` |
| + Completed uint64 `json:"completed,omitempty"` |
| + CompletedFailed bool `json:"completed_failed,omitempty"` |
| + Desired string `json:"desired,omitempty"` |
| + Reason string `json:"reason,omitempty"` |
| + Detail string `json:"detail,omitempty"` |
| + // Baking is the head a pending, unresolved request waits on (this box's |
| + // bake window). Reported beside Reason, never in place of it: Reason is |
| + // what the held wake check reads (Astra I1 #8). |
| + Baking string `json:"baking,omitempty"` |
| + At time.Time `json:"at"` |
| +} |
| + |
| +func ConsumedPath(optDir, host string) string { return filepath.Join(optDir, "consumed", host) } |
| + |
| +// ReadConsumed is (nil, nil) when root has never written one for host. |
| +func ReadConsumed(optDir, host string) (*Consumed, error) { |
| + if !controlname.ValidHostShape(host) { |
| + return nil, fmt.Errorf("consumed: invalid host %q", host) |
| + } |
| + raw, err := os.ReadFile(ConsumedPath(optDir, host)) |
| + if errors.Is(err, os.ErrNotExist) { |
| + return nil, nil |
| + } |
| + if err != nil { |
| + return nil, err |
| + } |
| + var c Consumed |
| + if err := json.Unmarshal(raw, &c); err != nil { |
| + return nil, fmt.Errorf("consumed %s: %v", host, err) |
| + } |
| + if c.Schema != 1 || c.Host != host { |
| + return nil, fmt.Errorf("consumed %s names %q, schema %d", host, c.Host, c.Schema) |
| + } |
| + return &c, nil |
| +} |
| + |
| +func WriteConsumed(optDir string, c Consumed) error { |
| + if !controlname.ValidHostShape(c.Host) { |
| + return fmt.Errorf("consumed: invalid host %q", c.Host) |
| + } |
| + c.Schema = 1 |
| + if c.At.IsZero() { |
| + c.At = time.Now().UTC().Truncate(time.Second) |
| + } |
| + raw, err := json.MarshalIndent(c, "", " ") |
| + if err != nil { |
| + return err |
| + } |
| + if err := os.MkdirAll(filepath.Dir(ConsumedPath(optDir, c.Host)), 0o755); err != nil { |
| + return err |
| + } |
| + return WriteFileAtomic(ConsumedPath(optDir, c.Host), raw, 0o644) |
| +} |
| + |
| +// Watermark is the highest completed request id either source knows (§4.3 |
| +// rule 2): an older local file on a box the instance returns to can never |
| +// make a newer, durably completed request look pending. |
| +func Watermark(c *Consumed, b *BuildState) uint64 { |
| + var w uint64 |
| + if c != nil { |
| + w = c.Completed |
| + } |
| + if b != nil && b.Completed > w { |
| + w = b.Completed |
| + } |
| + return w |
| +} |
| + |
| +// NextBuildState is the pre-exec write's merge (settings spec §4.2, §4.3 |
| +// rule 2): before build runs against the instance's database, last_executed |
| +// becomes build, root's resolution is carried, and the request completes if |
| +// build is its target. That write is what completes a request: at first |
| +// run, not at install. A request root recorded as failed is carried as |
| +// completed, failed. Schema, Host and At are the writer's. |
| +func NextBuildState(cur *BuildState, c *Consumed, build string) BuildState { |
| + next := BuildState{} |
| + if cur != nil { |
| + next = *cur |
| + next.Resolved = copyResolved(cur.Resolved) |
| + } |
| + if c != nil { |
| + if c.Resolved != nil && c.Resolved.ID > Watermark(nil, &next) { |
| + next.Resolved = copyResolved(c.Resolved) |
| + } |
| + switch { |
| + case c.Completed > next.Completed: |
| + next.Completed, next.CompletedFailed = c.Completed, c.CompletedFailed |
| + case c.Completed == next.Completed && c.CompletedFailed: |
| + // Root learned the completed request's T failed after it ran |
| + // (Astra I1 #10): the outcome changes, the request stays closed. |
| + next.CompletedFailed = true |
| + } |
| + } |
| + next.LastExecuted = build |
| + if next.Resolved != nil && next.Resolved.Target == build && next.Resolved.ID > next.Completed { |
| + next.Completed, next.CompletedFailed = next.Resolved.ID, false |
| + } |
| + return next |
| +} |
| + |
| +func copyResolved(r *ResolvedRequest) *ResolvedRequest { |
| + if r == nil { |
| + return nil |
| + } |
| + v := *r |
| + return &v |
| +} |
diff --git a/internal/release/buildstate_test.go b/internal/release/buildstate_test.go| new file mode 100644 |
| index 000000000..611c35f8b |
| --- /dev/null |
| +++ b/internal/release/buildstate_test.go |
| @@ -0,0 +1,152 @@ |
| +package release |
| + |
| +import ( |
| + "os" |
| + "path/filepath" |
| + "reflect" |
| + "testing" |
| + "time" |
| +) |
| + |
| +func TestBuildStateKeySitsBesideTheLock(t *testing.T) { |
| + if got := BuildStateKey("instances/a.example.com"); got != "instances/a.example.com/build-state.json" { |
| + t.Fatalf("BuildStateKey = %q", got) |
| + } |
| +} |
| + |
| +func TestConsumedRoundTrip(t *testing.T) { |
| + opt := t.TempDir() |
| + if c, err := ReadConsumed(opt, "a.example.com"); c != nil || err != nil { |
| + t.Fatalf("absent = %+v, %v", c, err) |
| + } |
| + // Reason is the plain string "request"; Task 2 gives it a named constant. |
| + want := Consumed{Schema: 1, Host: "a.example.com", Resolved: &ResolvedRequest{ID: 3, Target: "v2"}, Completed: 2, |
| + Desired: "v2", Reason: "request", At: time.Now().UTC().Truncate(time.Second)} |
| + if err := WriteConsumed(opt, want); err != nil { |
| + t.Fatal(err) |
| + } |
| + got, err := ReadConsumed(opt, "a.example.com") |
| + if err != nil || got.Resolved.Target != "v2" || got.Completed != 2 || got.Reason != "request" { |
| + t.Fatalf("round trip = %+v, %v", got, err) |
| + } |
| + if ConsumedPath(opt, "a.example.com") != filepath.Join(opt, "consumed", "a.example.com") { |
| + t.Fatal("the consumed file moved") |
| + } |
| +} |
| + |
| +func TestConsumedRefusesAnotherHostsFile(t *testing.T) { |
| + opt := t.TempDir() |
| + if err := WriteConsumed(opt, Consumed{Schema: 1, Host: "b.example.com"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := WriteFileAtomic(ConsumedPath(opt, "a.example.com"), mustRead(t, ConsumedPath(opt, "b.example.com")), 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := ReadConsumed(opt, "a.example.com"); err == nil { |
| + t.Fatal("a consumed file naming another host was accepted") |
| + } |
| +} |
| + |
| +func TestWatermarkIsTheHighestOfBoth(t *testing.T) { |
| + for _, c := range []struct { |
| + c *Consumed |
| + b *BuildState |
| + want uint64 |
| + }{ |
| + {nil, nil, 0}, |
| + {&Consumed{Completed: 4}, nil, 4}, |
| + {nil, &BuildState{Completed: 6}, 6}, |
| + {&Consumed{Completed: 4}, &BuildState{Completed: 6}, 6}, |
| + {&Consumed{Completed: 9}, &BuildState{Completed: 6}, 9}, |
| + } { |
| + if got := Watermark(c.c, c.b); got != c.want { |
| + t.Errorf("Watermark(%+v, %+v) = %d, want %d", c.c, c.b, got, c.want) |
| + } |
| + } |
| +} |
| + |
| +func TestBuildStateCopyColumn(t *testing.T) { |
| + c, err := ParseBuildStateCopy("") |
| + if err != nil || c.Known { |
| + t.Fatalf("an empty column is unknown: %+v %v", c, err) |
| + } |
| + in := BuildStateCopy{Known: true, Replica: true, State: &BuildState{Schema: 1, Host: "a.example.com", LastExecuted: "v1"}} |
| + out, err := ParseBuildStateCopy(in.Column()) |
| + if err != nil || !out.Known || !out.Replica || out.State.LastExecuted != "v1" { |
| + t.Fatalf("round trip = %+v %v", out, err) |
| + } |
| +} |
| + |
| +func TestNextBuildState(t *testing.T) { |
| + for _, tc := range []struct { |
| + name string |
| + cur *BuildState |
| + c *Consumed |
| + build string |
| + want BuildState |
| + }{ |
| + {"a: nil cur, nil consumed sets only LastExecuted", nil, nil, "v1", |
| + BuildState{LastExecuted: "v1"}}, |
| + {"b: a consumed resolution above the watermark is carried and completes when its target runs", |
| + &BuildState{Schema: 1, Host: "h", Completed: 2}, |
| + &Consumed{Resolved: &ResolvedRequest{ID: 3, Target: "v3"}, Completed: 2}, "v3", |
| + BuildState{Schema: 1, Host: "h", LastExecuted: "v3", Resolved: &ResolvedRequest{ID: 3, Target: "v3"}, Completed: 3}}, |
| + {"c: running another build carries the resolution but does not complete it", |
| + &BuildState{Schema: 1, Host: "h", Completed: 2}, |
| + &Consumed{Resolved: &ResolvedRequest{ID: 3, Target: "v3"}, Completed: 2}, "v2", |
| + BuildState{Schema: 1, Host: "h", LastExecuted: "v2", Resolved: &ResolvedRequest{ID: 3, Target: "v3"}, Completed: 2}}, |
| + {"d: a consumed resolution at or below the state's Completed is not carried", |
| + &BuildState{Schema: 1, Host: "h", Completed: 3}, |
| + &Consumed{Resolved: &ResolvedRequest{ID: 3, Target: "v3"}, Completed: 2}, "v3", |
| + BuildState{Schema: 1, Host: "h", LastExecuted: "v3", Completed: 3}}, |
| + {"e: a higher consumed Completed raises the state's and carries CompletedFailed", |
| + &BuildState{Schema: 1, Host: "h", Completed: 2}, |
| + &Consumed{Completed: 5, CompletedFailed: true}, "v1", |
| + BuildState{Schema: 1, Host: "h", LastExecuted: "v1", Completed: 5, CompletedFailed: true}}, |
| + {"f: an equal Completed with consumed CompletedFailed marks the state failed and does not reopen", |
| + &BuildState{Schema: 1, Host: "h", Completed: 4}, |
| + &Consumed{Completed: 4, CompletedFailed: true}, "v1", |
| + BuildState{Schema: 1, Host: "h", LastExecuted: "v1", Completed: 4, CompletedFailed: true}}, |
| + {"g: the state's own resolution from another box completes when its target runs", |
| + &BuildState{Schema: 1, Host: "h", Resolved: &ResolvedRequest{ID: 4, Target: "v3"}, Completed: 3}, |
| + nil, "v3", |
| + BuildState{Schema: 1, Host: "h", LastExecuted: "v3", Resolved: &ResolvedRequest{ID: 4, Target: "v3"}, Completed: 4}}, |
| + {"h: the watermark never decreases", |
| + &BuildState{Schema: 1, Host: "h", Completed: 6}, |
| + &Consumed{Completed: 4}, "v1", |
| + BuildState{Schema: 1, Host: "h", LastExecuted: "v1", Completed: 6}}, |
| + } { |
| + t.Run(tc.name, func(t *testing.T) { |
| + var curBefore BuildState |
| + if tc.cur != nil { |
| + curBefore = *tc.cur |
| + } |
| + got := NextBuildState(tc.cur, tc.c, tc.build) |
| + if !reflect.DeepEqual(got, tc.want) { |
| + t.Fatalf("NextBuildState = %+v (resolved %+v), want %+v (resolved %+v)", got, got.Resolved, tc.want, tc.want.Resolved) |
| + } |
| + if tc.cur != nil && !reflect.DeepEqual(*tc.cur, curBefore) { |
| + t.Fatalf("cur was mutated: %+v", *tc.cur) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestNextBuildStateDoesNotAliasConsumed(t *testing.T) { |
| + c := &Consumed{Resolved: &ResolvedRequest{ID: 3, Target: "v3"}} |
| + got := NextBuildState(nil, c, "v2") |
| + c.Resolved.Target = "changed" |
| + c.Resolved.ID = 99 |
| + if got.Resolved == nil || got.Resolved.Target != "v3" || got.Resolved.ID != 3 { |
| + t.Fatalf("the state aliases the consumed file: %+v", got.Resolved) |
| + } |
| +} |
| + |
| +func mustRead(t *testing.T, path string) []byte { |
| + t.Helper() |
| + b, err := os.ReadFile(path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return b |
| +} |