package versions import ( "context" "encoding/json" "errors" "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 git tags of the repository the name is built from, under prefix. Tags func(ctx context.Context, repository, prefix 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) } // FloorOf names what a floor is asked for: the name on its seat, and the repository and tag prefix of its git tags. type FloorOf struct{ Seat, Name, Repository, Prefix string } // Ask asks the three sources of a 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, of FloorOf) (Floor, map[string]string) { seat, name := of.Seat, of.Name 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, of.Repository, of.Prefix) } } 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, ", ")) } // The builder's state floor-waits (DLV-B4): each name waiting on its floor, keyed by its escaped name, so a restarted // builder still knows which names its silent sources' events wake; builder_floor_waits reads it. const ( FloorWaitsState = "floor-waits" FloorWaitsBucket = "mesh-delivery-builder_floor-waits" FloorWaitsTool = "builder_floor_waits" ) // FloorWaitKey is a wait's key in floor-waits. func FloorWaitKey(name string) string { return EscapeToken(name) } // FloorKeeper keeps the waits on the bus: Put a wait's value, Delete its key. type FloorKeeper interface { Put(key string, value []byte) error Delete(key string) error } // 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. Once Kept, every change is // written to the state floor-waits as well. type FloorWaits struct { mu sync.Mutex waits map[string]FloorWait keeper FloorKeeper // Logf says a wait that could not be kept; nil says nothing. Logf func(string, ...any) // Said is told each wait recorded (raised) and each answered (not raised): the builder's finding floor-silent. Said func(w FloorWait, raised bool) } // FloorSilent is the kind of the floor finding. const FloorSilent = "floor-silent" // Keep binds the state: the waits it holds that are not known here are taken in (a restarted builder's), every wait // known here is written to it, and every change from now on is. func (fw *FloorWaits) Keep(k FloorKeeper, held []FloorWait) { fw.mu.Lock() if fw.waits == nil { fw.waits = map[string]FloorWait{} } for _, w := range held { if _, had := fw.waits[w.Name]; !had && w.Name != "" { fw.waits[w.Name] = w } } fw.keeper = k all := make([]FloorWait, 0, len(fw.waits)) for _, w := range fw.waits { all = append(all, w) } fw.mu.Unlock() for _, w := range all { fw.put(k, w) } } func (fw *FloorWaits) put(k FloorKeeper, w FloorWait) { if k == nil { return } data, err := json.Marshal(w) if err == nil { err = k.Put(FloorWaitKey(w.Name), data) } if err != nil && fw.Logf != nil { fw.Logf("the floor wait of %s is not kept in %s (a restart forgets it until its next claim): %v", w.Name, FloorWaitsState, err) } } // DecodeFloorWait reads one value of floor-waits. func DecodeFloorWait(raw []byte) (FloorWait, error) { var w FloorWait if err := json.Unmarshal(raw, &w); err != nil { return FloorWait{}, err } if w.Name == "" { return FloorWait{}, errors.New("a floor wait names no name") } return w, nil } // 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() 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 k := fw.keeper fw.mu.Unlock() fw.put(k, w) if fw.Said != nil { fw.Said(w, true) } 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() w, had := fw.waits[name] delete(fw.waits, name) k := fw.keeper fw.mu.Unlock() if had && fw.Said != nil { fw.Said(w, false) } if had && k != nil { if err := k.Delete(FloorWaitKey(name)); err != nil && fw.Logf != nil { fw.Logf("the floor wait of %s is not cleared from %s: %v", name, FloorWaitsState, err) } } return had } // FloorWaitsAnswer is what builder_floor_waits answers: every name waiting on its floor, oldest first, with its finding. type FloorWaitsAnswer struct { Waits []FloorWaitRead `json:"waits"` } // FloorWaitRead is one wait as builder_floor_waits shows it. type FloorWaitRead struct { FloorWait Finding string `json:"finding"` } // AnswerFloorWaits is builder_floor_waits, read from the state floor-waits. func AnswerFloorWaits(r Reader) (FloorWaitsAnswer, error) { keys, err := r.Keys() if err != nil { return FloorWaitsAnswer{}, err } out := FloorWaitsAnswer{Waits: []FloorWaitRead{}} for _, k := range keys { raw, ok, err := r.Get(k) if err != nil { return FloorWaitsAnswer{}, err } if !ok { continue } w, err := DecodeFloorWait(raw) if err != nil { return FloorWaitsAnswer{}, fmt.Errorf("%s.%s: %w", FloorWaitsState, k, err) } out.Waits = append(out.Waits, FloorWaitRead{FloorWait: w, Finding: w.Finding()}) } sort.Slice(out.Waits, func(i, j int) bool { if !out.Waits[i].Since.Equal(out.Waits[j].Since) { return out.Waits[i].Since.Before(out.Waits[j].Since) } return out.Waits[i].Name < out.Waits[j].Name }) return out, nil } // FloorWaitsDescription is builder_floor_waits' description. const FloorWaitsDescription = "The names waiting on their floor (the builder's state floor-waits, design 51 *The floor*), oldest first: " + "each name, its registry seat, the sources still silent and why, since when, and the finding the builder says for it. " + "Nothing of a waiting name is numbered until all three sources answer; builder_floor_again asks one again by hand. Read-only." // 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 }