mesh-tools
b76cbf708a
mesh/merge-gate checking again: its repo-check failed at 12:59 UTC only on TestTheCataloguesDerivedSubjectsMeet, a catalogue fault (mesh-issues' dead subsc…
mesh/repo-check checking again: its repo-check failed at 12:59 UTC only on TestTheCataloguesDerivedSubjectsMeet, a catalogue fault (mesh-issues' dead subsc…
mesh/delivery superseded: a newer head of the same pull request
88 lines
3.0 KiB
Go
88 lines
3.0 KiB
Go
package main
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"github.com/novox/mesh-controller/internal/inventory"
|
|
)
|
|
|
|
// busWaitBound is how long a module whose ask the bus did not answer is asked again at every look before the plan fails
|
|
// (issue 1334).
|
|
const busWaitBound = 10 * time.Minute
|
|
|
|
// busDidNotAnswer says an ask failed because the bus's JetStream had no responder at that moment, as it has for a few seconds
|
|
// when the bus reloads its user list (novox/hq issue 353), and not because of the module: the same ask is made again.
|
|
func busDidNotAnswer(err error) bool {
|
|
if err == nil {
|
|
return false
|
|
}
|
|
if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, nats.ErrJetStreamNotEnabled) ||
|
|
errors.Is(err, jetstream.ErrJetStreamNotEnabled) || errors.Is(err, inventory.ErrNoChain) {
|
|
return true
|
|
}
|
|
// An error that crossed a boundary as text only.
|
|
text := err.Error()
|
|
return strings.Contains(text, "no responders available") || strings.Contains(text, "jetstream not enabled")
|
|
}
|
|
|
|
// waitForTheBus leaves a module whose ask the bus did not answer unasked, the plan building, and says so in the plan's note: the
|
|
// next look asks the tier's unasked modules again. It says false, so the caller fails the plan as for any refusal, once the bus
|
|
// has been silent for busWaitBound.
|
|
func waitForTheBus(p *inventory.Plan, state *inventory.PlanModule, name string, err error, now time.Time) bool {
|
|
if state.BusWaitFrom == nil {
|
|
state.BusWaitFrom = &now
|
|
}
|
|
if now.Sub(*state.BusWaitFrom) > busWaitBound {
|
|
return false
|
|
}
|
|
state.State, state.Why = "", ""
|
|
p.Note = fmt.Sprintf("tier %d: %s not asked: the bus did not answer (%v); asked again at the next look, failing after %s",
|
|
p.Tier, name, firstLine(err.Error()), busWaitBound)
|
|
return true
|
|
}
|
|
|
|
// busWaiting says an open plan has a module of its current tier that the bus did not answer an ask of: the look must come back
|
|
// for it, as nothing else will when no build of the tier is in flight (issue 1334).
|
|
func busWaiting(p inventory.Plan) bool {
|
|
return len(busWaitingIn(p)) > 0
|
|
}
|
|
|
|
// busWaitingIn are the modules of the plan's current tier left unasked because the bus did not answer, sorted.
|
|
func busWaitingIn(p inventory.Plan) []string {
|
|
if !p.Open() || p.Tier >= len(p.Tiers) {
|
|
return nil
|
|
}
|
|
var out []string
|
|
for _, m := range p.Tiers[p.Tier] {
|
|
if s := p.Modules[m]; s != nil && s.State == "" && s.BusWaitFrom != nil {
|
|
out = append(out, m)
|
|
}
|
|
}
|
|
sort.Strings(out)
|
|
return out
|
|
}
|
|
|
|
// busWaitWords is how a verb says modules are waiting for the bus: not asked yet, and asked again by the plan at its next look.
|
|
func busWaitWords(p inventory.Plan, modules []string) string {
|
|
return fmt.Sprintf("%s not asked yet: the bus did not answer; %s asks again at its next look", strings.Join(modules, ", "), p.ID)
|
|
}
|
|
|
|
// slicesWithout is names without those in drop, in order.
|
|
func slicesWithout(names, drop []string) []string {
|
|
var out []string
|
|
for _, n := range names {
|
|
if !slices.Contains(drop, n) {
|
|
out = append(out, n)
|
|
}
|
|
}
|
|
return out
|
|
}
|