package approval import ( "context" "encoding/json" "fmt" "strings" "time" "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/operatorask" "git.novox.be/novox/mesh-sdk/go/outward" ) // Wait is one mesh_wait call as the gate judges it (design 34 every-event §34 to §36): the event names given, the subjects // they became (after `from`), the filter and the timeout. type Wait struct { Names []string From string Subjects []string Filter map[string]any Timeout time.Duration } // Waited is a wait through the gate (every-event §34 to §36). An unmarked wait is heard at once by hear, given how long it // may wait. A marked one is heard only on the operator's warrant for that one wait, its timeout running from then; with // none it subscribes to nothing, its ask raised once under the runner's own module's name, and is answered as an error at // its timeout, when its caller goes, or as soon as the router's record says the ask was chosen, never in a hot loop. c is // the call as the console makes it: its address, its tool's name, its caller and its arguments as given. func (g *Gate) Waited(ctx context.Context, c Call, w Wait, hear func(within time.Duration) (any, error)) (any, error) { deadline := g.d.Now().Add(w.Timeout) // A wait from a seat is never marked, and a seat's event never is (§32, §34). if w.From == "seat" || !outsideSeats(w.Subjects) { return hear(w.Timeout) } if g.d.Catalog == nil { return nil, g.holdRefused(ctx, deadline, "this wait is refused: the tool runner reads no catalog half, so the marks are not read") } if err := g.d.Catalog.Ready(); err != nil { // Before the marks are read nothing is asked: the wait waits for the first view within its own timeout and is // then judged, and is refused at its timeout when it does not come (§34). if g.d.Wait != nil { err = g.waitCatalog(ctx, w.Timeout) } if err != nil { return nil, g.holdRefused(ctx, deadline, fmt.Sprintf("this wait is refused at its timeout: the marks are not yet read (%v); "+ "mesh_first_views says what the tool runner waits for", err)) } } if !overlapsAny(w.Subjects, g.waitMarks()) { return hear(deadline.Sub(g.d.Now())) } c.Module, c.Seat, c.Fingerprint = g.d.Runner, "", g.d.Build if c.Fingerprint == "" { return nil, g.holdRefused(ctx, deadline, "this wait needs the operator's approval, and the tool runner's own fingerprint is "+ "unknown (its MESH_FINGERPRINT was not set at its start), so no warrant can bind its build") } if b, ok := g.d.State.(Binding); g.d.State == nil || ok && !b.Bound() { return nil, g.holdRefused(ctx, deadline, g.refused(c, "", "the tool runner's membership binds no state "+StateName+" to keep its asks in").Error()) } if why := refusedUnasked(c.Args); why != "" { return nil, g.holdRefused(ctx, deadline, g.refused(c, "", why).Error()) } if why := g.held(); why != "" { return nil, g.holdRefused(ctx, deadline, g.refused(c, "", why).Error()) } act, err := ActOf(c, g.d.Node) if err != nil { return nil, g.holdRefused(ctx, deadline, g.refused(c, "", err.Error()).Error()) } digest, err := operatorask.ActDigest(act) if err != nil { return nil, g.holdRefused(ctx, deadline, g.refused(c, "", err.Error()).Error()) } named, err := g.d.State.Get(g.actKey(digest)) if err != nil { g.stateFailed(err) return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "its state "+StateName+" cannot be read: "+err.Error()).Error()) } g.stateRead() anew := func() (any, error) { id := newID() if err := g.keepAsk(c, act, digest, g.askForWait(c, w, digest, id)); err != nil { return nil, g.holdRefused(ctx, deadline, err.Error()) } return nil, g.holdAsk(ctx, deadline, c.Module, id, digest, true) } if named == nil { return anew() } var id string if err := json.Unmarshal(named.Value, &id); err != nil || id == "" { return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, fmt.Sprintf("its act's key in %s names no ask", StateName)).Error()) } rec, err := g.record(c.Module, id) switch { case err != nil: return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "the router's record of its ask "+id+" cannot be read: "+err.Error()).Error()) case rec == nil, rec.State == "open", rec.State == "held-back": return nil, g.holdAsk(ctx, deadline, c.Module, id, digest, false) case rec.State == "refused": return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "the router refused its ask "+id+": "+orNone(rec.Answer)).Error()) case rec.State != "chosen": return anew() // expired, cancelled, replaced: asked again (§15, §17) } wt := rec.Warrant if wt == nil { return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "the router's record of its ask "+id+" is chosen and carries no warrant").Error()) } at, _ := parseTime(wt.At) if wt.Option == OptionDecline { if g.d.Now().Before(at.Add(DeclinedFor)) { return nil, g.holdRefused(ctx, deadline, declinedWords(id, at)) } return anew() } if g.d.Now().After(at.Add(RunWithin)) { return anew() } var answer any var ran error // On the warrant it subscribes, and the wait's timeout runs from then (§35). h, err := g.d.Channel.Claim(c.Module, *wt, act, func() { answer, ran = hear(w.Timeout) }) switch { case h == ClaimDone: return anew() // already run: one warrant is one wait (§35) case h == ClaimRefused: return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "its warrant does not hold: "+errText(err)).Error()) case h == "" && err != nil: g.stateFailed(err) return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "its state "+StateName+" cannot be written: "+err.Error()).Error()) case err != nil: return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, err.Error()).Error()) } return g.ran(c, id, rec, answer, ran) } // outsideSeats says a subject is not a seat's: only such a subject can be marked (§32, §34). func outsideSeats(subjects []string) bool { for _, s := range subjects { if !strings.HasPrefix(s, "mesh.seat.") { return true } } return false } // waitMarks are the marked subjects of every module the catalog half lists (§33): the newest entry of every followed // branch, and the served entry of each module this runner serves (lead answer A-168), so a mark is never lost. func (g *Gate) waitMarks() map[string]string { m := Marks{Tools: map[string]string{}, Verbs: map[string]string{}, Subjects: map[string]string{}} defs := g.d.Catalog.Definitions() for _, e := range g.d.Catalog.Entries() { m.add(e, defs) } g.mu.Lock() served := make(map[string]string, len(g.served)) for module, fp := range g.served { served[module] = fp } g.mu.Unlock() for _, fp := range served { if e, ok, _ := g.d.Catalog.Served(fp); ok { m.add(e, defs) } } return m.Subjects } // overlapsAny says a subject a wait subscribes overlaps a marked subject (§34). func overlapsAny(subjects []string, marked map[string]string) bool { for _, s := range subjects { for mk := range marked { if buscompose.Overlaps(s, mk) { return true } } } return false } // waitCatalog waits for the catalog half's first view within the bound, or until the caller goes. func (g *Gate) waitCatalog(ctx context.Context, within time.Duration) error { done := make(chan error, 1) go func() { done <- g.d.Wait(within) }() select { case err := <-done: return err case <-ctx.Done(): return ctx.Err() } } // pause waits until the deadline or the caller's going. func (g *Gate) pause(ctx context.Context, deadline time.Time) { if left := deadline.Sub(g.d.Now()); left > 0 { select { case <-g.d.After(left): case <-ctx.Done(): } } } // holdRefused is a refused marked wait, said at its timeout or when its caller goes, never at once, so a watcher's loop // is never a hot one (§36). func (g *Gate) holdRefused(ctx context.Context, deadline time.Time, words string) error { g.pause(ctx, deadline) if !strings.HasPrefix(words, "Not run") { words = "Not run: " + words } return fmt.Errorf("%s", words) } // holdAsk holds a marked wait whose ask is open, subscribing to nothing but its ask's `decided` (§36): it answers, as an // error, at its timeout, when its caller goes, or once the router's record says the ask was chosen, the event only waking it. func (g *Gate) holdAsk(ctx context.Context, deadline time.Time, module, id, digest string, first bool) error { for { left := deadline.Sub(g.d.Now()) heard := false if left > 0 && ctx.Err() == nil { var err error if g.d.Hear != nil { heard, err = g.d.Hear(ctx, []string{g.d.Channel.Decided(module)}, left, func(body json.RawMessage) bool { var w struct { Ask string `json:"ask"` } return json.Unmarshal(body, &w) == nil && w.Ask == id }) } if g.d.Hear == nil || err != nil { g.pause(ctx, deadline) } } rec, _ := g.record(module, id) if heard && (rec == nil || rec.State != "chosen") { continue // the event only wakes it: the record says nothing chosen yet } return g.waitWords(rec, id, digest, first) } } // waitWords are §12's and §17's words for a marked wait not run, "this wait" for "the call" (§36). func (g *Gate) waitWords(rec *Record, id, digest string, first bool) error { switch { case rec == nil: return fmt.Errorf("Not run, asked; not yet taken by the router (ask %s, fingerprint %s); make this wait again later", id, Fingerprint(digest)) case rec.State == "chosen" && rec.Warrant != nil && rec.Warrant.Option == OptionDecline: at, _ := parseTime(rec.Warrant.At) return fmt.Errorf("%s", declinedWords(id, at)) case rec.State == "chosen", first: return fmt.Errorf("Not run: this wait needs the operator's approval. Asked on %s at %s (ask %s, fingerprint %s); make this wait "+ "again within 30 minutes of the approval; the ask expires at %s", orNone(strings.Join(rec.OfferedOn, ", ")), rec.Opened, id, Fingerprint(digest), rec.Expires) case rec.State == "open", rec.State == "held-back": return fmt.Errorf("Not run: still waiting for the operator's answer to ask %s (fingerprint %s); it expires at %s", id, Fingerprint(digest), rec.Expires) } return fmt.Errorf("Not run: the ask %s ended %s; make this wait again to ask anew", id, rec.State) } func declinedWords(id string, at time.Time) string { return fmt.Sprintf("Not run: the operator declined ask %s; this wait is not asked again until %s", id, formatTime(at.Add(DeclinedFor))) } // askForWait is a marked wait's ask (§35): "Wait on on ?", shortened as §10 says, and an explanation naming // every event name one per line, the filter, and the bound as the timeout in minutes, then the act's fingerprint. func (g *Gate) askForWait(c Call, w Wait, digest, id string) Ask { a := g.askFor(c, digest, id) filter, whole := "none", true if len(w.Filter) > 0 { raw, _ := json.Marshal(w.Filter) filter = string(raw) if r, ok := outward.Check(filter); !ok { filter, whole = shapeOf(r), false } } a.Headline = headlineOf("Wait on", strings.Join(w.Names, ", "), g.d.Node, whole) lines := append([]string{"Wait on " + g.d.Node + ", called through the mesh's tools, which cannot tell who called, for:"}, w.Names...) lines = append(lines, "Filter: "+filter, fmt.Sprintf("Timeout: %s minutes", minutes(w.Timeout)), "Fingerprint "+Fingerprint(digest)) a.Explanation = strings.Join(lines, "\n") a.Options[0].Does = "the same wait, made again within 30 minutes, subscribes once, its timeout running from then" a.Options[1].Does = "nothing is heard; the same wait is not asked again for 10 minutes" return a } // minutes is a timeout in minutes, as few digits as say it. func minutes(d time.Duration) string { return strings.TrimSuffix(strings.TrimRight(fmt.Sprintf("%.2f", d.Minutes()), "0"), ".") }