package witness import ( "context" "errors" "fmt" "strings" "time" "github.com/novox/mesh-host/internal/link" ) // Watcher judges every witnessed build on trial on this machine, every Every, until each is proved or // concluded (to-be 45 §8). // // **Time counts only while it could ask.** A build's bound is spent only on looks that got an answer // from the bus — the lease read, or the PING delivered — so a machine whose own link is down, or a // bus that refuses the question, never rolls a build back for the host's own trouble. A look that // could not ask counts towards GiveUp instead, and at GiveUp the build is said to be unwitnessed: // neither proved nor rolled back, and said. type Watcher struct { Root string // Node is this machine's node name, the instance its runtime answers PING as; Host its hostname, // as the controller's lease names its machine. Node, Host string // Asker is the link open now, or nil. Asker func() link.Asker Run Runner // Hold runs a change to what is placed on the machine in turn with every apply: the restoration // moves the directory an apply writes into. Hold func(func()) // Concluded is told every verdict, once, as it is reached: to say it now rather than at the next // report. Concluded func(link.Rollback) Say func(string) Now func() time.Time Every time.Duration } // Watch looks every Every until ctx ends. func (w *Watcher) Watch(ctx context.Context) { every := w.Every if every <= 0 { every = Every } ticker := time.NewTicker(every) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: w.Look(ctx) } } } // Look judges every build on trial once, and deletes every retired build no process runs from now. func (w *Watcher) Look(ctx context.Context) { for _, s := range Trials(w.Root) { w.look(ctx, s) } w.hold(func() { SweepAll(w.Root) }) } func (w *Watcher) look(ctx context.Context, s State) { every := w.Every if every <= 0 { every = Every } now := w.now() healthy, why, err := w.judge(ctx, s, now) w.hold(func() { // Read again in turn: an apply may have placed another build meanwhile, which starts its own // trial — this look was about the one before it. current, loadErr := Load(w.Root, s.Process) if loadErr != nil || current.Running != s.Running || !current.OnTrial() { return } switch { case err != nil: current.Unasked += every if current.Unasked < GiveUp { _ = Save(w.Root, current) return } v := link.Rollback{Component: Component(current.Witness), From: current.Running, Outcome: link.Unwitnessed, Why: fmt.Sprintf("its health could not be judged for %s: %v. The build before it is kept", GiveUp, err), At: now.UTC()} current.Verdicts = append(current.Verdicts, v) if Save(w.Root, current) == nil { w.concluded(v) } case healthy: if Proved(w.Root, s.Process) == nil { said := fmt.Sprintf("%s %s is healthy: %s; the build before it is retired", s.Process, short(s.Running), why) if draining, _ := Sweep(w.Root, s.Process); len(draining) > 0 { said += ", and deleted once nothing runs from it: " + strings.Join(draining, "; ") } else { said += " and deleted" } w.say(said) } default: current.Watched += every if current.Watched < current.Within { _ = Save(w.Root, current) return } if err := Save(w.Root, current); err != nil { return } bound := fmt.Sprintf("%s %s was not healthy within %s of starting: %s", s.Process, short(s.Running), current.Within, why) v, err := Restore(ctx, w.Root, s.Process, bound, w.Run, now) if err != nil { w.say(fmt.Sprintf("%s; and what was concluded could not be kept: %v", bound, err)) } w.concluded(v) } }) } // judge asks once. An error is a look that could not ask; otherwise healthy, and why not when not. func (w *Watcher) judge(ctx context.Context, s State, now time.Time) (bool, string, error) { var asker link.Asker if w.Asker != nil { asker = w.Asker() } if asker == nil { return false, "", errors.New("this host is not linked to the bus") } asking, cancel := context.WithTimeout(ctx, PingWithin) defer cancel() switch s.Witness { case ByLease: value, found, err := asker.ReadKey(asking, LeaseBucket, LeaseKey) if err != nil { return false, "", err } if !found { return false, "nobody holds the controller's lease", nil } lease, err := ParseLease(value) if err != nil { return false, err.Error(), nil } held, why := lease.HeldBySince(w.Host, s.Started, now) return held, why, nil case ByPing: err := asker.Ping(asking, s.Process, w.Node) switch { case err == nil: return true, "it answered PING", nil case errors.Is(err, link.ErrNoAnswer): return false, err.Error(), nil default: return false, "", err } } return false, "", fmt.Errorf("%s names no witness this host knows (%q)", s.Process, s.Witness) } func (w *Watcher) hold(f func()) { if w.Hold == nil { f() return } w.Hold(f) } func (w *Watcher) now() time.Time { if w.Now == nil { return time.Now() } return w.Now() } func (w *Watcher) say(line string) { if w.Say != nil { w.Say(line) } } func (w *Watcher) concluded(v link.Rollback) { w.say(fmt.Sprintf("%s %s: %s — %s", v.Component, v.Outcome, short(v.From), v.Why)) if w.Concluded != nil { w.Concluded(v) } }