diff --git a/modules/mesh-delivery-builder/internal/versions/floor.go b/modules/mesh-delivery-builder/internal/versions/floor.go new file mode 100644 index 00000000..d4b5dbd7 --- /dev/null +++ b/modules/mesh-delivery-builder/internal/versions/floor.go @@ -0,0 +1,179 @@ +package versions + +import ( + "context" + "fmt" + "slices" + "sort" + "strings" + "sync" + "time" +) + +// FloorSources are the three sources of a name's floor (design 51 *The floor*): each a tool call that answers the +// versions it knows of a name, or an error when it does not answer. +type FloorSources struct { + // Seat is `seat:.versions` of the registry seat the name is published to. + Seat func(ctx context.Context, seat, name string) ([]string, error) + // Tags is the forge bridge's bridge_tags: the package's git tags. + Tags func(ctx context.Context, name string) ([]string, error) + // Requirements is the indexer's indexer_requirements: the highest pseudo-versions consumers require. + Requirements func(ctx context.Context, name string) ([]string, error) +} + +// Ask asks the three sources of name; a source that errs, or is not wired, is named in Silent with its error in why. +func (s FloorSources) Ask(ctx context.Context, seat, name string) (Floor, map[string]string) { + var f Floor + why := map[string]string{} + ask := func(source string, call func() ([]string, error), into *[]string) { + if call == nil { + f.Silent, why[source] = append(f.Silent, source), "not asked: the builder has no path to it" + return + } + got, err := call() + if err != nil { + f.Silent, why[source] = append(f.Silent, source), err.Error() + return + } + *into = got + } + var seatCall, tagsCall, reqCall func() ([]string, error) + if s.Seat != nil { + seatCall = func() ([]string, error) { return s.Seat(ctx, seat, name) } + } + if s.Tags != nil { + tagsCall = func() ([]string, error) { return s.Tags(ctx, name) } + } + if s.Requirements != nil { + reqCall = func() ([]string, error) { return s.Requirements(ctx, name) } + } + ask(SourceSeat, seatCall, &f.Seat) + ask(SourceTags, tagsCall, &f.Tags) + ask(SourceRequirements, reqCall, &f.Requirements) + return f, why +} + +// What wakes a name waiting on its floor, never a timer (design 51, DLV-B3b): the source that was silent saying +// something, or its module's health condition clearing. +const ( + // WakePublished is the holder's `published` on the builder's durable consumer of `.events`. + WakePublished = "published" + // WakeForge is the forge bridge's next message on its stream `forge`. + WakeForge = "forge" + // WakeAccepted is the indexer's next `manifest-accepted`. + WakeAccepted = "manifest-accepted" + // WakeCleared is the condition keeper's clear of a silent module's health condition on `mesh-conditions`. + WakeCleared = "condition-cleared" + // WakeByHand is builder_floor_again, a hand act. + WakeByHand = "by hand" +) + +// sourceOfWake is the floor source each event says something of; a clear names the source by its module. +var sourceOfWake = map[string]string{WakePublished: SourceSeat, WakeForge: SourceTags, WakeAccepted: SourceRequirements} + +// ModuleOfSource is the module whose health condition's clear wakes a silent source. +var ModuleOfSource = map[string]string{ + SourceTags: "mesh-delivery-forge-bridge", + SourceRequirements: "mesh-delivery-indexer", +} + +// FloorWait is one name waiting on its floor: the sources still silent and why, since when; the finding the builder +// writes names them, cleared once all three answer. +type FloorWait struct { + Name string `json:"name"` + Seat string `json:"seat"` + Silent []string `json:"silent"` + Why map[string]string `json:"why,omitempty"` + Since time.Time `json:"since"` + // Holder is the module holding the name's seat: its condition's clear wakes the seat source. + Holder string `json:"holder,omitempty"` +} + +// Finding is the floor finding's words. +func (w FloorWait) Finding() string { + var parts []string + for _, s := range w.Silent { + p := s + if why := w.Why[s]; why != "" { + p += " (" + why + ")" + } + parts = append(parts, p) + } + return fmt.Sprintf("the floor of %s is not known since %s: %s did not answer; nothing of it is numbered until all three answer", + w.Name, w.Since.UTC().Format(time.RFC3339), strings.Join(parts, ", ")) +} + +// FloorWaits are the names waiting on their floor. Wake answers which names to ask again for an event, Answered drops +// a name once all three answered; the builder's finding for each is the wait's Finding. +type FloorWaits struct { + mu sync.Mutex + waits map[string]FloorWait +} + +// Wait records that name's floor is not known. +func (fw *FloorWaits) Wait(name, seat, holder string, f Floor, why map[string]string, now time.Time) FloorWait { + fw.mu.Lock() + defer fw.mu.Unlock() + if fw.waits == nil { + fw.waits = map[string]FloorWait{} + } + w, had := fw.waits[name] + if !had { + w.Since = now + } + w.Name, w.Seat, w.Holder, w.Silent, w.Why = name, seat, holder, slices.Clone(f.Silent), why + fw.waits[name] = w + return w +} + +// Answered drops name: its floor is known. It says whether the name waited, so its finding is cleared. +func (fw *FloorWaits) Answered(name string) bool { + fw.mu.Lock() + defer fw.mu.Unlock() + _, had := fw.waits[name] + delete(fw.waits, name) + return had +} + +// Wake answers the names to ask their floor again on an event: kind one of the Wake constants; seat the seat a +// `published` came on; module the module whose condition cleared. A name is woken only when the event says something +// of a source it waits on; WakeByHand with a name wakes that name alone. +func (fw *FloorWaits) Wake(kind, seat, module, name string) []string { + fw.mu.Lock() + defer fw.mu.Unlock() + var out []string + for n, w := range fw.waits { + switch kind { + case WakeByHand: + if n == name { + out = append(out, n) + } + case WakeCleared: + for _, s := range w.Silent { + if ModuleOfSource[s] == module || (s == SourceSeat && w.Holder != "" && w.Holder == module) { + out = append(out, n) + break + } + } + default: + src := sourceOfWake[kind] + if slices.Contains(w.Silent, src) && (kind != WakePublished || seat == "" || seat == w.Seat) { + out = append(out, n) + } + } + } + sort.Strings(out) + return out +} + +// Waiting is every wait, by name. +func (fw *FloorWaits) Waiting() []FloorWait { + fw.mu.Lock() + defer fw.mu.Unlock() + out := make([]FloorWait, 0, len(fw.waits)) + for _, w := range fw.waits { + out = append(out, w) + } + sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name }) + return out +} diff --git a/modules/mesh-delivery-builder/internal/versions/oldform.go b/modules/mesh-delivery-builder/internal/versions/oldform.go new file mode 100644 index 00000000..c05e45f6 --- /dev/null +++ b/modules/mesh-delivery-builder/internal/versions/oldform.go @@ -0,0 +1,165 @@ +package versions + +import ( + "slices" + "sort" + "time" +) + +// The old reference form's window (design 51 *Every build keeps the old reference form*, ADR 0624, ADR 0618 §2-§4, +// DLV-B9). The builder writes into every build key each reference form a counted node-engine states, the oldest of them +// into the key's single reference and the others beside it; the new form only once every present node-engine states it. +// The window closes when no counted node-engine states the old form alone. What each node states is read through the +// node-engine seat's verb `manifest-reads` at the builder's start and again on the seat's event +// `manifest-reads-changed`, never by a watch of the node-engine's state and never on a timer. + +// NodeReads is one counted node: what its node-engine states of the reference forms, and whether it is present. Stated +// nil is a node-engine that states nothing, or does not serve `manifest-reads` yet: read as the baseline's forms. +type NodeReads struct { + Node string `json:"node"` + Present bool `json:"present"` + Stated []string `json:"stated,omitempty"` + // Serves is false while the node-engine does not serve `manifest-reads`: it counts as stating the baseline. + Serves bool `json:"serves"` +} + +// FormsAnswer is what the builder writes into a build key, and why. +type FormsAnswer struct { + // Single is the form written into the key's single reference: the oldest a counted engine states. + Single string `json:"single"` + // Beside are the other forms written beside it, oldest first. + Beside []string `json:"beside,omitempty"` + // Open is whether the old form's window is open: a counted engine states the oldest form alone. + Open bool `json:"open"` + // Holding are the counted nodes whose engine states the oldest form alone (present or away), sorted. + Holding []string `json:"holding,omitempty"` +} + +// Forms decides the reference forms of new build keys. forms are every reference form the builder can write, oldest +// first; baseline what an engine stating nothing reads; dropped the nodes a person dropped from the count by hand +// (builder_old_form, a hand act), which then count no more. A node leaves the count when it is retired (by hand here), +// never because it is away (ADR 0624 §1). +func Forms(forms, baseline []string, nodes []NodeReads, dropped map[string]bool) FormsAnswer { + if len(forms) == 0 { + return FormsAnswer{} + } + oldest, newest := forms[0], forms[len(forms)-1] + type counted struct { + node string + reads []string + } + var cs []counted + everyPresentNewest := true + var holding []string + for _, n := range nodes { + if dropped[n.Node] { + continue + } + stated := n.Stated + if !n.Serves || len(stated) == 0 { + stated = baseline + } + var reads []string + for _, f := range forms { + if slices.Contains(stated, f) { + reads = append(reads, f) + } + } + if n.Present && !slices.Contains(reads, newest) { + everyPresentNewest = false + } + if len(reads) == 1 && reads[0] == oldest { + holding = append(holding, n.Node) + } + cs = append(cs, counted{n.Node, reads}) + } + // The new form is written once every present engine states it (ADR 0618 §2); each counted engine then gets the + // newest form it reads among those written, so a form leaves once no counted engine needs it (ADR 0624 §1). An + // engine reading none of them is held by the assigner (ADR 0624 §2), not served by an older form here. + keep := map[string]bool{} + if everyPresentNewest { + keep[newest] = true + } + for _, c := range cs { + for i := len(c.reads) - 1; i >= 0; i-- { + if f := c.reads[i]; f != newest || everyPresentNewest { + keep[f] = true + break + } + } + } + var kept []string + for _, f := range forms { + if keep[f] { + kept = append(kept, f) + } + } + if len(kept) == 0 { + kept = []string{oldest} // nothing counted: the oldest stays, so no engine is cut off + } + sort.Strings(holding) + var beside []string + if len(kept) > 1 { + beside = kept[1:] + } + return FormsAnswer{Single: kept[0], Beside: beside, Open: len(holding) > 0, Holding: holding} +} + +// OldFormWindow is the window's record: whether it is open, who holds it and since when, and the nodes dropped by +// hand; what builder_old_form answers and what `old-form-closed` carries when it closes. +type OldFormWindow struct { + Open bool `json:"open"` + Holding []string `json:"holding,omitempty"` + Since map[string]time.Time `json:"since,omitempty"` + Dropped []Dropped `json:"dropped,omitempty"` + Closed time.Time `json:"closed,omitempty"` + // LastHolding are the nodes that held it last, said when it closes. + LastHolding []string `json:"last_holding,omitempty"` +} + +// Dropped is one node dropped from the count by hand: a hand act with who and why. +type Dropped struct { + Node string `json:"node"` + By string `json:"by"` + Why string `json:"why"` + At time.Time `json:"at"` +} + +// FindingAfter is how long a node holds the window open before it is the builder's finding naming the node. +const FindingAfter = 7 * 24 * time.Hour + +// Step moves the window to what Forms answered at now: it answers the new window, whether it closed in this step (so +// the builder says `old-form-closed` and its build-log line once), and the nodes holding it past FindingAfter (the +// finding, cleared when the window closes). +func (w OldFormWindow) Step(a FormsAnswer, now time.Time) (next OldFormWindow, closed bool, overdue []string) { + next = OldFormWindow{Dropped: w.Dropped, Since: map[string]time.Time{}} + if !a.Open { + if w.Open { + next.Closed, next.LastHolding = now, w.Holding + return next, true, nil + } + next.Closed, next.LastHolding = w.Closed, w.LastHolding + return next, false, nil + } + next.Open, next.Holding = true, a.Holding + for _, n := range a.Holding { + since, had := w.Since[n] + if !had { + since = now + } + next.Since[n] = since + if now.Sub(since) >= FindingAfter { + overdue = append(overdue, n) + } + } + return next, false, overdue +} + +// DroppedSet is the dropped nodes as Forms takes them. +func (w OldFormWindow) DroppedSet() map[string]bool { + out := map[string]bool{} + for _, d := range w.Dropped { + out[d.Node] = true + } + return out +} diff --git a/modules/mesh-delivery-builder/internal/versions/versions_test.go b/modules/mesh-delivery-builder/internal/versions/versions_test.go index 2ab59a1b..e6a90e8c 100644 --- a/modules/mesh-delivery-builder/internal/versions/versions_test.go +++ b/modules/mesh-delivery-builder/internal/versions/versions_test.go @@ -3,6 +3,8 @@ package versions import ( "context" "encoding/json" + "errors" + "reflect" "strings" "sync" "testing" @@ -225,6 +227,98 @@ func TestTheBuildersVersions(t *testing.T) { } } }) + t.Run("the floor wakes on its silent source, never a timer", func(t *testing.T) { + ctx := context.Background() + down := func(context.Context, string) ([]string, error) { return nil, errors.New("no answer") } + src := FloorSources{ + Seat: func(context.Context, string, string) ([]string, error) { return []string{"0.1.3"}, nil }, + Tags: down, + Requirements: func(context.Context, string) ([]string, error) { return nil, nil }, + } + f, why := src.Ask(ctx, "oci-artifact-registry", "m/program") + if f.Known() || !reflect.DeepEqual(f.Silent, []string{SourceTags}) || why[SourceTags] != "no answer" { + t.Fatalf("Ask = %+v, %v; want bridge_tags silent", f, why) + } + var fw FloorWaits + w := fw.Wait("m/program", "oci-artifact-registry", "distribution", f, why, time.Unix(0, 0)) + if !strings.Contains(w.Finding(), "bridge_tags (no answer)") { + t.Errorf("finding = %q", w.Finding()) + } + for _, c := range []struct { + kind, seat, module, name string + want []string + }{ + {WakePublished, "oci-artifact-registry", "", "", nil}, // the seat answered already: a publish wakes nothing + {WakeAccepted, "", "", "", nil}, + {WakeForge, "", "", "", []string{"m/program"}}, + {WakeCleared, "", "mesh-delivery-forge-bridge", "", []string{"m/program"}}, // recovered with nothing to say + {WakeCleared, "", "mesh-delivery-indexer", "", nil}, + {WakeByHand, "", "", "m/program", []string{"m/program"}}, + {WakeByHand, "", "", "other", nil}, + } { + if got := fw.Wake(c.kind, c.seat, c.module, c.name); !reflect.DeepEqual(got, c.want) { + t.Errorf("Wake(%s %s %s %s) = %v, want %v", c.kind, c.seat, c.module, c.name, got, c.want) + } + } + seatDown := Floor{Silent: []string{SourceSeat}} + fw.Wait("distribution/program", "container-image-registry", "distribution", seatDown, nil, time.Unix(0, 0)) + if got := fw.Wake(WakePublished, "container-image-registry", "", ""); !reflect.DeepEqual(got, []string{"distribution/program"}) { + t.Errorf("the holder's published woke %v, want its own image's name", got) + } + if got := fw.Wake(WakeCleared, "", "distribution", ""); !reflect.DeepEqual(got, []string{"distribution/program"}) { + t.Errorf("the holder's clear woke %v", got) + } + if !fw.Answered("m/program") || fw.Answered("m/program") || len(fw.Waiting()) != 1 { + t.Error("Answered drops the wait once") + } + }) + t.Run("old form", func(t *testing.T) { + forms := []string{"ref:artifact-store", "ref:registry"} + base := []string{"ref:artifact-store"} + both := []string{"ref:artifact-store", "ref:registry"} + for _, c := range []struct { + name string + nodes []NodeReads + dropped map[string]bool + want FormsAnswer + }{ + {"one present node lacking it keeps the old form alone", + []NodeReads{{Node: "a", Present: true, Serves: true, Stated: both}, {Node: "b", Present: true, Serves: true, Stated: base}}, nil, + FormsAnswer{Single: "ref:artifact-store", Open: true, Holding: []string{"b"}}}, + {"a node not serving manifest-reads counts as the baseline", + []NodeReads{{Node: "a", Present: true, Serves: true, Stated: both}, {Node: "b", Present: true, Stated: both}}, nil, + FormsAnswer{Single: "ref:artifact-store", Open: true, Holding: []string{"b"}}}, + {"every present node states it: the new form beside the old, an away old node keeping the old in the single reference", + []NodeReads{{Node: "a", Present: true, Serves: true, Stated: both}, {Node: "c", Serves: true, Stated: base}}, nil, + FormsAnswer{Single: "ref:artifact-store", Beside: []string{"ref:registry"}, Open: true, Holding: []string{"c"}}}, + {"the away node back with the new engine closes the window", + []NodeReads{{Node: "a", Present: true, Serves: true, Stated: both}, {Node: "c", Present: true, Serves: true, Stated: []string{"ref:registry"}}}, nil, + FormsAnswer{Single: "ref:registry"}}, + {"a retired node dropped by hand counts no more", + []NodeReads{{Node: "a", Present: true, Serves: true, Stated: []string{"ref:registry"}}, {Node: "c", Serves: true, Stated: base}}, map[string]bool{"c": true}, + FormsAnswer{Single: "ref:registry"}}, + } { + if got := Forms(forms, base, c.nodes, c.dropped); !reflect.DeepEqual(got, c.want) { + t.Errorf("%s: Forms = %+v, want %+v", c.name, got, c.want) + } + } + t0 := time.Unix(0, 0) + w, closed, overdue := OldFormWindow{}.Step(FormsAnswer{Open: true, Holding: []string{"c"}}, t0) + if closed || len(overdue) != 0 || !w.Open { + t.Fatalf("Step = %+v %v %v", w, closed, overdue) + } + w, _, overdue = w.Step(FormsAnswer{Open: true, Holding: []string{"c"}}, t0.Add(FindingAfter)) + if !reflect.DeepEqual(overdue, []string{"c"}) { + t.Errorf("seven days held: overdue %v, want c", overdue) + } + w, closed, _ = w.Step(FormsAnswer{}, t0.Add(FindingAfter+time.Hour)) + if !closed || w.Open || !reflect.DeepEqual(w.LastHolding, []string{"c"}) { + t.Errorf("closing: %+v closed=%v", w, closed) + } + if _, again, _ := w.Step(FormsAnswer{}, t0.Add(FindingAfter+2*time.Hour)); again { + t.Error("a closed window closes again: old-form-closed would be said twice") + } + }) t.Run("fill", func(t *testing.T) { if _, full := Fill(Cap*79/100, Cap); full { t.Error("79 % is full")