package process import ( "context" "encoding/json" "errors" "fmt" "strings" "time" "github.com/nats-io/nats.go/jetstream" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/approval" "git.novox.be/novox/mesh-sdk/go/busretry" "git.novox.be/novox/mesh-sdk/go/seatdef" ) // Gating is what a holder's gate reaches beyond this connection (design 47 G §677; lead answers A-219, A-224, A-228): the // operator channel (mesh-contracts' operatorchannelclient.GateChannel over this connection, reading the router's record // under the module's own name, §700) and the holder's held finding writer. type Gating struct { Channel approval.Channel Findings approval.FindingWriter // Build is the holder's own fingerprint; empty reads it from the process's environment (OwnBuild). Build string Logf func(string, ...any) } // Gate is the holder's gate over this connection (§677): its state `approvals`, its calls of the seat `mesh-hand-acts`'s // `warranted` to record a run, and its seat's `gated-ended`, each on the process's own account. AnswerSeat holds the // seat's marked verbs through it (Gated); the process hears its decided with HearDecided and reads it at its start with // the gate's StartHolding. func (c *Conn) Gate(g Gating) *approval.Gate { build := g.Build if build == "" { build = OwnBuild().Fingerprint } state := c.State(approval.StateName) return approval.New(approval.Deps{Node: c.account.Node, Runner: c.account.Module, Holding: true, Build: build, State: state, Bounds: heldBounds{state}, Channel: g.Channel, Findings: g.Findings, Logf: g.Logf, Warranted: func(asker, ask string, open func() (bool, error)) error { _, err := c.CallKeptWhileOpen(busretry.Open{Key: ask, Open: func(string) (bool, error) { return open() }, Logf: g.Logf}, "seat:mesh-hand-acts.warranted", map[string]string{"asker": asker, "ask": ask}) return err }, Say: func(subject string, body any, id string) error { return c.emitOn(subject, body, []stdio.EmitOptions{{MessageID: id}}) }}) } // heldBounds is the holder's bound keys in its own state approvals (lead answer A-255): a key with its own time to live, // whose end the bus says as a marker that a watch hears, so a waiting call's 60 minutes need no clock in the process. type heldBounds struct{ s State } func (b heldBounds) CreateTTL(key string, value any, ttl time.Duration) error { _, err := b.s.CreateTTL(key, value, ttl) return err } func (b heldBounds) Delete(key string) error { return b.s.Delete(key) } // Watch hands expired each key under pattern whose own time to live ended, heard from now on: the bus's marker of a time // to live is a purge, which this state's own writes never make (a delete is not one). func (b heldBounds) Watch(pattern string, expired func(key string)) (func(), error) { ctx, cancel := context.WithCancel(context.Background()) bind, stop := context.WithTimeout(ctx, stateTimeout) kv, err := b.s.store(bind) stop() if err != nil { cancel() return nil, err } w, err := kv.Watch(ctx, watchPattern(pattern), jetstream.UpdatesOnly()) if err != nil { cancel() return nil, fmt.Errorf("%s's state %s, watching %q: %w", b.s.Module, b.s.Name, pattern, err) } go func() { defer cancel() defer func() { _ = w.Stop() }() for e := range w.Updates() { if e != nil && e.Operation() == jetstream.KeyValuePurge { expired(e.Key()) } } }() return cancel, nil } // HearDecided hands the gate each `decided` of the holder's own asks that hub hears (§677): only the ask's id is read // from it, the gate reading the router's own record. func HearDecided(g *approval.Gate, ch approval.Channel, module string, hub *EventHub) error { return hub.Handle(ch.Decided(module), func(m jetstream.Msg) { var w struct { Ask string `json:"ask"` } if err := json.Unmarshal(m.Data(), &w); err != nil || w.Ask == "" { _ = m.Term() return } g.Decided(w.Ask) _ = m.Ack() }) } // HeldWithin bounds a caller's wait on a held call's end: the ask's 60 minutes and the 30 a run may come after its choice. var HeldWithin = approval.AskLasts + approval.RunWithin // awaitGated waits on the held call's own end on its seat's gated-ended, from the last said on its subject (so an end said // before the wait began is read), and is its answer: `{answer, approval}` when it ran (lead answer A-158), its words as // the call's error otherwise. Every failure is an answer, never asked again: the call reached its holder (ADR 0668 §2). func (c *Conn) awaitGated(subject string, held approval.Held) (json.RawMessage, error) { seat, ok := seatOfSubject(subject) if !ok { return nil, answered(fmt.Sprintf("%s answered held, and is no seat's subject", subject)) } js, err := jetstream.New(c.nc) if err != nil { return nil, answered(err.Error()) } on := approval.GatedEndedSubject(seat, c.account.User, held.Call) stream := seatdef.KeptStream(seat, seatdef.Message{Name: approval.GatedEndedEvent}) ctx, cancel := context.WithTimeout(context.Background(), HeldWithin) defer cancel() cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: []string{on}, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy}) if err != nil { return nil, answered(fmt.Sprintf("the call %s waits on ask %s, and its end on %s cannot be heard: %v; %s.approvals reads it", held.Call, held.Asked, stream, err, seat)) } for { m, err := cons.Next(jetstream.FetchMaxWait(time.Minute)) if err == nil { _ = m.Ack() return endOf(m.Data()) } if ctx.Err() != nil || !errors.Is(err, jetstream.ErrNoMessages) && !errors.Is(err, context.DeadlineExceeded) { return nil, answered(fmt.Sprintf("the call %s waits on ask %s, and no end of it was heard (%v); %s.approvals reads it", held.Call, held.Asked, err, seat)) } } } // endOf is a gated-ended's body as the call's answer. func endOf(data []byte) (json.RawMessage, error) { var e approval.GatedEnded if err := json.Unmarshal(data, &e); err != nil { return nil, answered("the end of the call is not read: " + err.Error()) } if e.Outcome != approval.OutcomeRan || e.Error != "" { words := e.Error if words == "" { words = "the call ended " + e.Outcome } return nil, answered(fmt.Sprintf("Not run: %s (ask %s, call %s)", words, e.Ask, e.Call)) } return json.Marshal(approval.Answered{Answer: e.Answer, Approval: e.Approval}) } // seatOfSubject is the seat a call subject `mesh.seat..…` names. func seatOfSubject(subject string) (string, bool) { rest, ok := strings.CutPrefix(subject, "mesh.seat.") if !ok { return "", false } seat, _, ok := strings.Cut(rest, ".") return seat, ok && seat != "" }