package approval import ( "encoding/json" "errors" "fmt" "strings" "time" "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" seat "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/keys" "git.novox.be/novox/mesh-sdk/go/views" ) // The bounds of a run's recording (§13, §16; lead answers A-157, A-158). const ( UnansweredAfter = 5 * time.Minute KeptFor = 48 * time.Hour SaidBefore = time.Hour ) // Bounds is the runner's key-ttl state `bound` (lead answers A-157, A-158): a key created with its own time to live, which // the bus removes when it passes, every watch hearing the removal as an expiry, so the gate's waits need no timer. type Bounds interface { CreateTTL(key string, value any, ttl time.Duration) error Delete(key string) error // Watch hands expired every key under the pattern whose time to live passed, until the stop is called. Watch(pattern string, expired func(key string)) (stop func(), err error) } // Answered is a call run on the operator's approval: the tool's own answer and the line naming the approval (§16), in the // form a process holder's `gated-ended` says it (design 47 G §677; lead answer A-158). type Answered struct { Answer any `json:"answer"` Approval string `json:"approval"` } // The bound keys of a run (lead answer A-158; §13). func (g *Gate) unansweredKey(id string) string { return g.d.Node + ".unanswered." + id } func (g *Gate) unrecordedKey(id string) string { return g.d.Node + ".unrecorded." + id } // ran is a call run on the operator's approval: answered with the tool's answer and the approval's line, and its warrant // recorded in the hand-act log (§16). func (g *Gate) ran(c Call, id string, rec *seat.AsksByAskerIDValue, answer any, err error) (any, error) { channel := "" at := "" if w := rec.Warrant; w != nil { channel, at = w.Channel, w.At } if channel == "" { channel = strings.Join(rec.OfferedOn, ", ") } line := fmt.Sprintf("run on the operator's approval via %s at %s, ask %s", orNone(channel), at, id) g.d.Logf("[mesh-tools] %s on %s %s", c.Address, g.d.Node, line) g.bindRecording(c.Module, id) g.d.Go(func() { g.recordRun(c.Module, id) }) if err != nil { return nil, fmt.Errorf("%w (%s)", err, line) } return Answered{Answer: answer, Approval: line}, nil } // bindRecording keeps the run's two bound keys: one that raises `call-unanswered` when `warranted` has not answered within // five minutes, one that says an unrecorded run an hour before its key's 48 hours end (§13, §16). func (g *Gate) bindRecording(module, id string) { if g.d.Bounds == nil { return } for key, ttl := range map[string]time.Duration{g.unansweredKey(id): UnansweredAfter, g.unrecordedKey(id): KeptFor - SaidBefore} { if err := g.d.Bounds.CreateTTL(key, map[string]string{"asker": module, "ask": id}, ttl); err != nil { g.d.Logf("[mesh-tools] the bound %s of the run of ask %s is not kept: %v", key, id, err) } } } // recordRun calls `mesh-hand-acts.warranted` for one acted ask, each try a kept call open while the ask lacks `recorded` // (§16, A-157), and on its answer writes `recorded` and clears what waited on it. One recording per ask at a time. func (g *Gate) recordRun(module, id string) { if !g.recording(id, true) { return } defer g.recording(id, false) if g.d.Warranted == nil { g.d.Logf("[mesh-tools] the run of ask %s is not recorded: the tool runner reaches no hand-act log", id) return } err := g.d.Warranted(module, id, func() (bool, error) { return g.unrecorded(id) }) g.judged(id) // made again: its answer, or the bound of its key, judges it from now if err != nil { g.d.Logf("[mesh-tools] mesh-hand-acts.warranted has not answered for ask %s (%v): asked again when the hand-act seat's "+ "holder is heard again and at the next start", id, err) return } if err := g.markRecorded(id); err != nil { g.d.Logf("[mesh-tools] ask %s is recorded in the hand-act log and its mark recorded is not kept: %v", id, err) return } if g.d.Bounds != nil { _ = g.d.Bounds.Delete(g.unansweredKey(id)) _ = g.d.Bounds.Delete(g.unrecordedKey(id)) } g.finding(id, KindCallUnanswered, false, "mesh-hand-acts.warranted answered for ask "+id, "") } // recording claims or releases an ask's recording: false when one runs already. func (g *Gate) recording(id string, start bool) bool { g.mu.Lock() defer g.mu.Unlock() if g.inFlight == nil { g.inFlight = map[string]bool{} } if start { if g.inFlight[id] { return false } g.inFlight[id] = true return true } delete(g.inFlight, id) return true } // kept is the ask the client keeps under its id, with its revision; nil when there is none. func (g *Gate) kept(id string) (*operatorchannelclient.Kept, uint64, error) { e, err := g.d.State.Get(g.askKey(id)) if err != nil || e == nil { return nil, 0, err } var k operatorchannelclient.Kept if err := json.Unmarshal(e.Value, &k); err != nil { return nil, 0, err } return &k, e.Revision, nil } // unrecorded says the ask is acted on and its run not yet recorded: the key that keeps `warranted` open (§16). func (g *Gate) unrecorded(id string) (bool, error) { k, _, err := g.kept(id) if err != nil || k == nil { return false, err } return k.Acted && !k.Recorded, nil } // markRecorded writes `recorded` on the kept ask by compare-and-set, again when another write came first. func (g *Gate) markRecorded(id string) error { for { k, rev, err := g.kept(id) if err != nil || k == nil { return fmt.Errorf("the kept ask cannot be read: %v", err) } if k.Recorded { return nil } k.Recorded = true if _, err := g.d.State.Update(g.askKey(id), k, rev); err == nil { return nil } else if !isChanged(err) { return err } } } // Recover calls `warranted` again for every acted, unrecorded ask this machine's keys hold within their 48 hours (§16): at // each start, and when the hand-act seat's holder is heard again (A-157). func (g *Gate) Recover() { if g.d.State == nil { return } keys, err := g.d.State.Keys() if err != nil { g.d.Logf("[mesh-tools] the acted, unrecorded asks cannot be read (%v): read again at the next start or holder heard", err) return } prefix := g.d.Node + ".ask." for _, key := range keys { if !strings.HasPrefix(key, prefix) { continue } id := strings.TrimPrefix(key, prefix) k, _, err := g.kept(id) if err != nil || k == nil || !k.Acted || k.Recorded || g.d.Now().Sub(k.Asked) > KeptFor { continue } module := moduleOf(k) g.pend(id) g.d.Go(func() { g.recordRun(module, id) }) } } // moduleOf is the module an ask was asked under: the warrant's asker. func moduleOf(k *operatorchannelclient.Kept) string { if k.Answer != nil { return k.Answer.Asker } return "" } // WatchBounds hears the run's bound keys expire (A-157): `call-unanswered` raised when `warranted` has not answered within // five minutes; an unrecorded run said in the log an hour before its key's 48 hours end (§13). func (g *Gate) WatchBounds() (func(), error) { if g.d.Bounds == nil { return func() {}, nil } return g.d.Bounds.Watch(g.d.Node+".**", func(key string) { if key == g.unreadableKey() { g.unreadableExpired() return } rest := strings.TrimPrefix(key, g.d.Node+".") kind, id, _ := strings.Cut(rest, ".") if open, err := g.unrecorded(id); err != nil || !open { return } switch kind { case "unanswered": g.finding(id, KindCallUnanswered, true, fmt.Sprintf("mesh-hand-acts.warranted has not answered for ask %s, run on the "+ "operator's approval on %s, within %s: the run is not yet in the hand-act log", id, g.d.Node, UnansweredAfter), "a run's hand-act is unrecorded on "+g.d.Node) case "unrecorded": g.d.Logf("[mesh-tools] ERROR the run of ask %s on %s is still unrecorded an hour before its key's 48 hours end", id, g.d.Node) } }) } // Pender is the runner's held writer's pending (views.HeldWriter's Pend and Judged): the keys it cannot judge yet since its // start, named in its `findings-stated` (design 47 G §384). type Pender interface { Pend(views.PendingKey) Judged(key string) } // pend names an acted, unrecorded ask's `call-unanswered` in the runner's `findings-stated` until its `warranted` is made // again (§16; design 47 G §384), its bound the five minutes the finding waits. func (g *Gate) pend(id string) { if p, ok := g.d.Findings.(Pender); ok { p.Pend(views.PendingKey{Key: g.unansweredFinding(id), Since: g.d.Now().UTC(), Bound: UnansweredAfter.String()}) } } // judged ends that pending. func (g *Gate) judged(id string) { if p, ok := g.d.Findings.(Pender); ok { p.Judged(g.unansweredFinding(id)) } } // unansweredFinding is the key of an ask's `call-unanswered` in the runner's state findings (lead answer A-154). func (g *Gate) unansweredFinding(id string) string { return views.FindingKey(scopeMachine, keys.EscapeName(g.d.Node+"."+id), KindCallUnanswered) } // isChanged says a compare-and-set lost to another write. func isChanged(err error) bool { return errors.Is(err, stdio.ErrChanged) }