jschoubben
7033d5e80d
An ask's asker is the last token of its accept subject; the holder checks the kind against the callers the build seat's definition lists (Puller.Callers, CallersFromDefinitions), refuses an ask of another kind answering error naming the kind and the asker and ending it failed, takes only a build and a run on main's subject, and gives back an ask the seat list does not judge yet. A second worker per queue (AskNamedWorker, filtered to accept.<queue>.*) is pulled after main's; the holder says `named worker not bound` while its membership lists none. Builds: design 47 G §543, §544, §481, A-163, A-170 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
98 lines
3.0 KiB
Go
98 lines
3.0 KiB
Go
package steprun
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"time"
|
|
|
|
sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
)
|
|
|
|
// SeatAsks is a build seat's workers as a tools bundle reaches them: through its runtime's seat pull and settle
|
|
// (stdio.SeatPull, SeatSettle; issue 1887 A11), one message per pull, only from the workers the module's membership
|
|
// lists.
|
|
type SeatAsks struct {
|
|
Language string
|
|
// Pull and Settle are stdio.SeatPull and stdio.SeatSettle when nil; a test gives its own.
|
|
Pull func(worker string, wait time.Duration) (*stdio.Pulled, error)
|
|
Settle func(id, fate string) error
|
|
// SettleAfter is stdio.SeatSettleAfter when nil (ADR 0664 §1).
|
|
SettleAfter func(id string, delay time.Duration) error
|
|
// Logf says `named worker not bound` once for each queue whose named worker the membership does not list yet.
|
|
Logf func(format string, args ...any)
|
|
|
|
mu sync.Mutex
|
|
noted map[string]bool
|
|
}
|
|
|
|
// Next is AskSource's.
|
|
func (s *SeatAsks) Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error) {
|
|
pull, settle := s.Pull, s.Settle
|
|
if pull == nil {
|
|
pull = stdio.SeatPull
|
|
}
|
|
if settle == nil {
|
|
settle = stdio.SeatSettle
|
|
}
|
|
worker, err := sdkstep.AskWorker(s.Language, kind)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
named, err := sdkstep.AskNamedWorker(s.Language, kind)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
// Main's worker first, then the named one (design 47 G §544): a holder of G pulls main's alone until its membership
|
|
// names the named worker, and says so meanwhile. A wait is shared by the two.
|
|
half := wait / 2
|
|
p, err := pull(worker, min(wait, 0))
|
|
if err == nil && p == nil {
|
|
var nerr error
|
|
if p, nerr = pull(named, 0); nerr != nil {
|
|
s.noteUnbound(kind, nerr)
|
|
p = nil
|
|
} else {
|
|
s.mu.Lock()
|
|
delete(s.noted, kind)
|
|
s.mu.Unlock()
|
|
}
|
|
}
|
|
if err == nil && p == nil && wait > 0 {
|
|
if p, err = pull(worker, half); err == nil && p == nil {
|
|
if p, err = pull(named, wait-half); err != nil {
|
|
s.noteUnbound(kind, err)
|
|
p, err = nil, nil
|
|
}
|
|
}
|
|
}
|
|
if err != nil || p == nil {
|
|
return nil, err
|
|
}
|
|
after := s.SettleAfter
|
|
if after == nil {
|
|
after = stdio.SeatSettleAfter
|
|
}
|
|
fate := func(f string) func() error { return func() error { return settle(p.ID, f) } }
|
|
return &Pulled{Data: p.Body, Subject: p.Subject, Delivered: p.Delivered, Ack: fate(stdio.SettleAck), InProgress: fate(stdio.SettleProgress),
|
|
Nak: fate(stdio.SettleNak), Term: fate(stdio.SettleTerm),
|
|
NakAfter: func(delay time.Duration) error { return after(p.ID, delay) }}, nil
|
|
}
|
|
|
|
// noteUnbound says `named worker not bound` for a queue, once until it is bound (design 47 G §544).
|
|
func (s *SeatAsks) noteUnbound(kind string, err error) {
|
|
s.mu.Lock()
|
|
if s.noted == nil {
|
|
s.noted = map[string]bool{}
|
|
}
|
|
said := s.noted[kind]
|
|
s.noted[kind] = true
|
|
s.mu.Unlock()
|
|
if !said && s.Logf != nil {
|
|
s.Logf("named worker not bound: the %s queue's named asks wait until the membership lists %s: %v", kind, "its named worker", err)
|
|
}
|
|
}
|