DLV-B3, DLV-B3b, DLV-B9: the floor's three sources and what wakes a silent one; the old reference form's window
The floor asks the seat's versions, bridge_tags and indexer_requirements; a silent source is named in the floor finding and woken by an event of that source (the holder's published, the forge bridge's next message, the indexer's next manifest-accepted) or its module's condition clearing, or by hand (builder_floor_again), never by a timer (design 51 *The floor*, ADR 0617 §1). The old form's window (ADR 0624 §1-§3, ADR 0618 §2): each counted engine gets the newest form it reads among those written, the new form only once every present engine states it, the oldest kept in the single reference; a node not serving manifest-reads counts as the baseline; a node dropped by hand counts no more; old-form-closed is said once, and a node holding it seven days is a finding. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -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:<registry>.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 `<seat>.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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user