Add the seat artifact-store, the seat-holders state with its reader, and the builder's log layout, so the store's sweep, who holds which seat and build history leave the controller (issue 1955, ADR 0571 clauses 1, 2, 4)
This commit is contained in:
@@ -56,6 +56,44 @@ func TestTheControlRecordsReadTheirWritersValues(t *testing.T) {
|
||||
if _, err := EncodeNodeHealth(NodeHealth{}); err == nil {
|
||||
t.Error("a report naming no node was encoded")
|
||||
}
|
||||
// ADR 0571 clause 4: who holds which seat, from placement; a node seat by its node.
|
||||
held := SeatHoldersOf(map[string]Placement{
|
||||
"ace": {Node: "ace", Holds: []Holding{{Claim: "mesh-dns-resolver", Scope: "mesh", Module: "dnsmasq"}, {Claim: "node-files", Scope: "node", Module: "mesh-files"}}},
|
||||
"novox": {Node: "novox", Holds: []Holding{{Claim: "mesh-dns-resolver", Scope: "mesh", Module: "dnsmasq"}, {Claim: "mesh-broker", Scope: "mesh", Module: "nats"}}},
|
||||
})
|
||||
keys := map[string][]byte{}
|
||||
for k, v := range held {
|
||||
keys[k], _ = json.Marshal(v)
|
||||
}
|
||||
if h, err := HoldersOf(keys, "mesh-dns-resolver", ""); err != nil || len(h) != 2 || h[0].Node != "ace" {
|
||||
t.Errorf("a replicated seat's holders are %v, %v", h, err)
|
||||
}
|
||||
if h, _ := HoldersOf(keys, "node-files", "ace"); len(h) != 1 || h[0].Module != "mesh-files" || len(held) != 3 {
|
||||
t.Errorf("a node seat's holder is %v, keys %v", h, held)
|
||||
}
|
||||
if h, _ := HoldersOf(keys, "node-files", "novox"); h != nil {
|
||||
t.Errorf("a node seat nobody holds there is held by %v", h)
|
||||
}
|
||||
// ADR 0571 clause 1: the builder's log, a start and an end, and the duration between them.
|
||||
at := time.Date(2026, 10, 29, 22, 0, 0, 0, time.UTC)
|
||||
start := BuildLogEntry{Event: BuildStartedEvent, Module: "letta", FP: "src2:a", Source: "abc", Node: "ace", At: at}
|
||||
end := start
|
||||
end.Event, end.At, end.Outcome, end.Artifacts = BuildEndedEvent, at.Add(3*time.Minute), "done", []BuiltDigest{{Name: "image", Digest: "sha256:1"}}
|
||||
for _, e := range []BuildLogEntry{start, end} {
|
||||
raw, err := EncodeBuildLogEntry(e)
|
||||
back, derr := DecodeBuildLogEntry(raw)
|
||||
if err != nil || derr != nil || back.Event != e.Event || back.Outcome != e.Outcome {
|
||||
t.Errorf("%s read back as %+v, %v %v", e.Event, back, err, derr)
|
||||
}
|
||||
}
|
||||
if d, ok := BuildDuration(start, end); !ok || d != 3*time.Minute {
|
||||
t.Errorf("the build took %s, %v", d, ok)
|
||||
}
|
||||
bad := end
|
||||
bad.Outcome = ""
|
||||
if _, err := EncodeBuildLogEntry(bad); err == nil {
|
||||
t.Error("an end with no outcome was encoded")
|
||||
}
|
||||
if k, err := HeldValueKey("novox/mesh-catalog#42", "novox", "mail", "quota"); k != "novox.mesh-catalog.42.novox.mail.quota" || err != nil {
|
||||
t.Errorf("a held value's key is %q, %v", k, err)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,179 @@
|
||||
package delivery
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"time"
|
||||
)
|
||||
|
||||
// **Who holds which seat, and the builder's log** (novox/hq ADR 0571 clauses 1 and 4, issue 1955).
|
||||
|
||||
// SeatHoldersState is the state the seat mesh-placement's holder writes of who holds which seat on which node (clause 4):
|
||||
// one key per seat held once for the mesh, `<seat>`, and one per node of a seat held on every node, `<seat>.<node>`.
|
||||
const (
|
||||
SeatHoldersState = "seat-holders"
|
||||
SeatHoldersBucket = AssignerModule + "_" + SeatHoldersState
|
||||
SeatHoldersRead = AssignerModule + "." + SeatHoldersState
|
||||
)
|
||||
|
||||
// SeatHolder is one module holding a seat on a node.
|
||||
type SeatHolder struct {
|
||||
Module string `json:"module"`
|
||||
Node string `json:"node"`
|
||||
}
|
||||
|
||||
// SeatHolders is one key of seat-holders: the seat, its scope, the node for a node seat's key, and its holders
|
||||
// (several only for a replicated mesh seat), sorted.
|
||||
type SeatHolders struct {
|
||||
V int `json:"v"`
|
||||
Seat string `json:"seat"`
|
||||
Scope string `json:"scope"`
|
||||
Node string `json:"node,omitempty"`
|
||||
Holders []SeatHolder `json:"holders"`
|
||||
}
|
||||
|
||||
// SeatHoldersKey is a seat's key: `<seat>` for a seat held once for the mesh (node empty), `<seat>.<node>` for a node
|
||||
// seat on one node.
|
||||
func SeatHoldersKey(seat, node string) (string, error) {
|
||||
if err := checkName("seat", seat); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if node == "" {
|
||||
return seat, nil
|
||||
}
|
||||
if err := checkName("node", node); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return seat + "." + node, nil
|
||||
}
|
||||
|
||||
// SeatHoldersOf is the state composed from every node's placement (ADR 0567 §1): each holding by the seat it claims,
|
||||
// keyed as SeatHoldersKey says, with the module holding it and its node. Deterministic, so a writer writes only a key
|
||||
// that differs.
|
||||
func SeatHoldersOf(placements map[string]Placement) map[string]SeatHolders {
|
||||
out := map[string]SeatHolders{}
|
||||
nodes := make([]string, 0, len(placements))
|
||||
for n := range placements {
|
||||
nodes = append(nodes, n)
|
||||
}
|
||||
sort.Strings(nodes)
|
||||
for _, n := range nodes {
|
||||
for _, h := range placements[n].Holds {
|
||||
node := ""
|
||||
if h.Scope == "node" {
|
||||
node = n
|
||||
}
|
||||
key, err := SeatHoldersKey(h.Claim, node)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
sh := out[key]
|
||||
sh.V, sh.Seat, sh.Scope, sh.Node = Version, h.Claim, h.Scope, node
|
||||
sh.Holders = append(sh.Holders, SeatHolder{Module: h.Module, Node: n})
|
||||
out[key] = sh
|
||||
}
|
||||
}
|
||||
for k, sh := range out {
|
||||
sort.Slice(sh.Holders, func(i, j int) bool {
|
||||
return sh.Holders[i].Node+"/"+sh.Holders[i].Module < sh.Holders[j].Node+"/"+sh.Holders[j].Module
|
||||
})
|
||||
out[k] = sh
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// HoldersOf reads who holds seat (on node, for a node seat) from the state's keys as a reader watched them: the node's
|
||||
// key first, then the seat's. None when nobody holds it.
|
||||
func HoldersOf(keys map[string][]byte, seat, node string) ([]SeatHolder, error) {
|
||||
for _, n := range []string{node, ""} {
|
||||
key, err := SeatHoldersKey(seat, n)
|
||||
if err != nil {
|
||||
if n == "" {
|
||||
return nil, err
|
||||
}
|
||||
continue
|
||||
}
|
||||
raw, ok := keys[key]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
var sh SeatHolders
|
||||
if err := json.Unmarshal(raw, &sh); err != nil {
|
||||
return nil, fmt.Errorf("delivery: the holders of %s are not readable: %w", key, err)
|
||||
}
|
||||
if sh.V != Version {
|
||||
return nil, fmt.Errorf("delivery: the holders of %s are of version %d, and this reads %d", key, sh.V, Version)
|
||||
}
|
||||
return sh.Holders, nil
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// The builder's log (clause 1): a stream of its own, `log`, one entry when a build of a fingerprint starts and one when
|
||||
// it ends, kept 30 days.
|
||||
const (
|
||||
BuilderLogStream = "log"
|
||||
BuildStartedEvent = "build.started"
|
||||
BuildEndedEvent = "build.ended"
|
||||
// BuilderLogDays is how long the log keeps its entries.
|
||||
BuilderLogDays = 30
|
||||
)
|
||||
|
||||
// BuiltDigest is one artifact a build ended with, by its name and digest.
|
||||
type BuiltDigest struct {
|
||||
Name string `json:"name"`
|
||||
Digest string `json:"digest"`
|
||||
}
|
||||
|
||||
// BuildLogEntry is one entry of the builder's log: the module, the fingerprint, the source revision, the build node and
|
||||
// the time; at the end, the outcome and each artifact's digest.
|
||||
type BuildLogEntry struct {
|
||||
V int `json:"v"`
|
||||
Event string `json:"event"`
|
||||
Module string `json:"module"`
|
||||
FP string `json:"fp"`
|
||||
Source string `json:"source"`
|
||||
Node string `json:"node"`
|
||||
At time.Time `json:"at"`
|
||||
Outcome string `json:"outcome,omitempty"`
|
||||
Artifacts []BuiltDigest `json:"artifacts,omitempty"`
|
||||
}
|
||||
|
||||
// EncodeBuildLogEntry is e as JSON at Version; it refuses an entry a reader could not place.
|
||||
func EncodeBuildLogEntry(e BuildLogEntry) ([]byte, error) {
|
||||
e.V = Version
|
||||
switch {
|
||||
case e.Event != BuildStartedEvent && e.Event != BuildEndedEvent:
|
||||
return nil, fmt.Errorf("delivery: a build log entry is %s or %s, not %q", BuildStartedEvent, BuildEndedEvent, e.Event)
|
||||
case e.Module == "" || e.FP == "" || e.Node == "" || e.At.IsZero():
|
||||
return nil, fmt.Errorf("delivery: a build log entry names its module, fingerprint, node and time")
|
||||
case e.Event == BuildEndedEvent && e.Outcome == "":
|
||||
return nil, fmt.Errorf("delivery: a build's end names its outcome")
|
||||
case e.Event == BuildStartedEvent && (e.Outcome != "" || len(e.Artifacts) > 0):
|
||||
return nil, fmt.Errorf("delivery: a build's start names no outcome and no artifact")
|
||||
}
|
||||
return json.Marshal(e)
|
||||
}
|
||||
|
||||
// DecodeBuildLogEntry reads one entry.
|
||||
func DecodeBuildLogEntry(data []byte) (BuildLogEntry, error) {
|
||||
var e BuildLogEntry
|
||||
if err := json.Unmarshal(data, &e); err != nil {
|
||||
return BuildLogEntry{}, fmt.Errorf("delivery: a build log entry is not readable: %w", err)
|
||||
}
|
||||
if e.V != Version {
|
||||
return BuildLogEntry{}, fmt.Errorf("delivery: a build log entry of version %d, and this reads %d", e.V, Version)
|
||||
}
|
||||
return e, nil
|
||||
}
|
||||
|
||||
// BuildDuration is a build's duration, its end less its start (clause 1): false unless both are the same module's
|
||||
// build of the same fingerprint, the start first.
|
||||
func BuildDuration(start, end BuildLogEntry) (time.Duration, bool) {
|
||||
if start.Event != BuildStartedEvent || end.Event != BuildEndedEvent || start.Module != end.Module || start.FP != end.FP ||
|
||||
end.At.Before(start.At) {
|
||||
return 0, false
|
||||
}
|
||||
return end.At.Sub(start.At), true
|
||||
}
|
||||
Reference in New Issue
Block a user