diff --git a/node-tools/internal/bus/wait.go b/node-tools/internal/bus/wait.go index 4dda03b..aa2e961 100644 --- a/node-tools/internal/bus/wait.go +++ b/node-tools/internal/bus/wait.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "slices" "strings" "time" @@ -23,31 +24,42 @@ type WaitIssued struct { Subject string `json:"subject"` } -// Heard is one event a wait heard: the subject it arrived on, its body and when it arrived. +// Heard is one event a wait heard: the subject it arrived on, its body, when it arrived, and when it was published, the +// event's own header `x-time` as its emitter wrote it, empty for an event that carries none (design 34 every-event §12). type Heard struct { - Subject string - Body json.RawMessage - At time.Time + Subject string + Body json.RawMessage + At time.Time + Published string } // HearFirst subscribes the subjects for this call alone and answers the first message whose body match // accepts (every one, when match is nil), or ok false once `within` has passed. Every subscription is -// closed before it returns. A subscription the bus refuses is said as an error naming the subject, at once; -// a caller gone — ctx done — ends the wait at once with ctx's error, so nothing waits on for nobody. -func (c *Conn) HearFirst(ctx context.Context, subjects []string, within time.Duration, match func(json.RawMessage) bool) (heard Heard, ok bool, err error) { +// closed before it returns. A subscription the bus refuses is noted and the others go on waiting; refused lists +// those noted by the time it answers (design 34 every-event §16). When the bus refused every subscription of the +// call, it still waits out `within` and then answers the refusal, so a caller looping on it asks once a timeout and +// never at once again. A caller gone — ctx done — ends the wait at once with ctx's error, so nothing waits on for +// nobody. +func (c *Conn) HearFirst(ctx context.Context, subjects []string, within time.Duration, match func(json.RawMessage) bool) (heard Heard, ok bool, refused []string, err error) { if len(subjects) == 0 { - return Heard{}, false, fmt.Errorf("a wait names at least one subject") + return Heard{}, false, nil, fmt.Errorf("a wait names at least one subject") + } + var unique []string + for _, subject := range subjects { + if !slices.Contains(unique, subject) { + unique = append(unique, subject) + } } got := make(chan Heard, 1) - refused := make(chan string, len(subjects)) + refusals := make(chan string, len(unique)) var subs []*nats.Subscription defer func() { for _, s := range subs { _ = s.Unsubscribe() } c.mu.Lock() - for _, subject := range subjects { - delete(c.waits[subject], refused) + for _, subject := range unique { + delete(c.waits[subject], refusals) if len(c.waits[subject]) == 0 { delete(c.waits, subject) } @@ -58,14 +70,14 @@ func (c *Conn) HearFirst(ctx context.Context, subjects []string, within time.Dur if c.waits == nil { c.waits = map[string]map[chan string]bool{} } - for _, subject := range subjects { + for _, subject := range unique { if c.waits[subject] == nil { c.waits[subject] = map[chan string]bool{} } - c.waits[subject][refused] = true + c.waits[subject][refusals] = true } c.mu.Unlock() - for _, subject := range subjects { + for _, subject := range unique { sub, err := c.nc.Subscribe(subject, func(msg *nats.Msg) { body := json.RawMessage(msg.Data) if !json.Valid(body) { @@ -76,31 +88,39 @@ func (c *Conn) HearFirst(ctx context.Context, subjects []string, within time.Dur return } select { - case got <- Heard{Subject: msg.Subject, Body: body, At: time.Now().UTC()}: + case got <- Heard{Subject: msg.Subject, Body: body, At: time.Now().UTC(), Published: msg.Header.Get("x-time")}: default: } }) if err != nil { - return Heard{}, false, fmt.Errorf("could not wait on %s: %w", subject, err) + return Heard{}, false, nil, fmt.Errorf("could not wait on %s: %w", subject, err) } subs = append(subs, sub) } // The bus has every subscription — or has said which it refused — once this returns. if err := c.nc.FlushTimeout(5 * time.Second); err != nil { - return Heard{}, false, fmt.Errorf("the bus did not take the wait: %w", err) + return Heard{}, false, nil, fmt.Errorf("the bus did not take the wait: %w", err) } timer := time.NewTimer(within) defer timer.Stop() - select { - case h := <-got: - return h, true, nil - case subject := <-refused: - return Heard{}, false, fmt.Errorf("the bus refused this account a subscription on %s: it is not in the "+ - "account's wait grant (novox/hq ADR 0367 §2)", subject) - case <-timer.C: - return Heard{}, false, nil - case <-ctx.Done(): - return Heard{}, false, ctx.Err() + for { + select { + case h := <-got: + return h, true, refused, nil + case subject := <-refusals: + if !slices.Contains(refused, subject) { + refused = append(refused, subject) + } + case <-timer.C: + if len(refused) == len(unique) { + return Heard{}, false, refused, fmt.Errorf("the bus refused this account a subscription on %s: the "+ + "grants part grants this machine's tool runner every event subject (novox/hq ADR 0670 §1), so its "+ + "grant is not composed yet; read the state from its owner meanwhile", strings.Join(refused, ", ")) + } + return Heard{}, false, refused, nil + case <-ctx.Done(): + return Heard{}, false, refused, ctx.Err() + } } } diff --git a/node-tools/internal/bus/wait_test.go b/node-tools/internal/bus/wait_test.go index 45d3b5a..c12d8d7 100644 --- a/node-tools/internal/bus/wait_test.go +++ b/node-tools/internal/bus/wait_test.go @@ -2,22 +2,24 @@ package bus_test import ( "context" - "encoding/json" "errors" "strings" "testing" "time" "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" "github.com/novox/mesh-tools/node-tools/internal/bus" mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" ) -// novox/hq ADR 0367 §2, §5: the bus refuses the tool runner's account any subject its wait grant does not -// name — another asker's `decided` among them — and a wait on one says the bus refused it, at once. +// novox/hq ADR 0670 §1 (design 34 every-event §12, §16): a subscription the bus refuses is noted and the others go on +// waiting, a heard event carries its own time; a wait whose every subscription is refused answers only after its +// timeout, in words naming the grants part's grant of every event subject, never a wait grant (E-2). func TestTheBusRefusesAWaitOutsideTheGrant(t *testing.T) { granted := "mesh.seat.operator-channel.event.decided.claude-code" + other := "mesh.seat.operator-channel.event.decided.shop" s, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, NoLog: true, NoSigs: true, Users: []*server.User{{Username: "desk.node-tools", Password: "pw", Permissions: &server.Permissions{ Publish: &server.SubjectPermission{Allow: []string{"$JS.API.DIRECT.GET.>", "mesh.>"}}, @@ -33,33 +35,65 @@ func TestTheBusRefusesAWaitOutsideTheGrant(t *testing.T) { t.Cleanup(s.Shutdown) c := dial(t, s.ClientURL(), "node-tools", "desk", "desk.node-tools") + // Every subscription refused: answered after the timeout, never at once, in §16's words. start := time.Now() - _, ok, err := c.HearFirst(context.Background(), []string{"mesh.seat.operator-channel.event.decided.shop"}, 5*time.Second, nil) - if ok || err == nil || !strings.Contains(err.Error(), "refused") { - t.Errorf("a wait on another asker's decided: heard %v, %v", ok, err) + _, ok, refused, err := c.HearFirst(context.Background(), []string{other}, 300*time.Millisecond, nil) + if took := time.Since(start); took < 300*time.Millisecond || took > 3*time.Second { + t.Errorf("a wait refused whole answered after %s, want its timeout of 300ms", took) } - if time.Since(start) > 3*time.Second { - t.Errorf("the refusal took %s to be said", time.Since(start)) + if ok || err == nil || !strings.Contains(err.Error(), "the bus refused this account a subscription on "+other) || + !strings.Contains(err.Error(), "every event subject (novox/hq ADR 0670 §1)") || + strings.Contains(err.Error(), "wait grant") || strings.Contains(err.Error(), "0367") || + len(refused) != 1 || refused[0] != other { + t.Errorf("a wait on another asker's decided: heard %v, refused %v, %v", ok, refused, err) } - done := make(chan bus.Heard, 1) - go func() { - h, ok, err := c.HearFirst(context.Background(), []string{granted}, 5*time.Second, func(body json.RawMessage) bool { return true }) - if ok && err == nil { - done <- h - } - }() - time.Sleep(300 * time.Millisecond) - if err := c.Say(granted, []byte(`{"ask":"a1"}`)); err != nil { + // One refused beside one granted: the granted one hears, with the event's own time, and the refused one is listed. + pub, err := nats.Connect(s.ClientURL(), nats.UserInfo("desk.node-tools", "pw")) + if err != nil { t.Fatal(err) } - select { - case h := <-done: - if h.Subject != granted || string(h.Body) != `{"ask":"a1"}` { - t.Errorf("heard %+v", h) + t.Cleanup(pub.Close) + for _, published := range []string{"2026-11-12T03:00:00.5Z", ""} { + type result struct { + h bus.Heard + refused []string + err error + } + done := make(chan result, 1) + go func() { + h, ok, refused, err := c.HearFirst(context.Background(), []string{other, granted}, 5*time.Second, nil) + if !ok && err == nil { + err = errors.New("timed out") + } + done <- result{h, refused, err} + }() + // The event is said again until the wait, subscribed in its own time, hears it. + say := time.NewTicker(20 * time.Millisecond) + var r result + heard: + for { + select { + case r = <-done: + break heard + case <-say.C: + msg := nats.NewMsg(granted) + msg.Data = []byte(`{"ask":"a1"}`) + if published != "" { + msg.Header.Set("x-time", published) + } + if err := pub.PublishMsg(msg); err != nil { + t.Fatal(err) + } + } + } + say.Stop() + if r.err != nil || r.h.Subject != granted || string(r.h.Body) != `{"ask":"a1"}` || r.h.Published != published { + t.Errorf("heard %+v, %v; want published %q", r.h, r.err, published) + } + if len(r.refused) > 1 || len(r.refused) == 1 && r.refused[0] != other { + t.Errorf("refused %v, want at most %s", r.refused, other) } - case <-time.After(5 * time.Second): - t.Fatal("a wait on the granted subject heard nothing") } } @@ -70,7 +104,7 @@ func TestAWaitEndsWhenItsCallerIsGone(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) time.AfterFunc(200*time.Millisecond, cancel) start := time.Now() - _, ok, err := c.HearFirst(ctx, []string{"mesh.mod.mesh-knowledge.event.reviewed"}, time.Minute, nil) + _, ok, _, err := c.HearFirst(ctx, []string{"mesh.mod.mesh-knowledge.event.reviewed"}, time.Minute, nil) if ok || !errors.Is(err, context.Canceled) { t.Errorf("a cancelled wait answered %v, %v", ok, err) } diff --git a/node-tools/internal/console/wait.go b/node-tools/internal/console/wait.go index 9ee1a41..1194b5b 100644 --- a/node-tools/internal/console/wait.go +++ b/node-tools/internal/console/wait.go @@ -86,18 +86,25 @@ func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[strin family[subject] = name } - heard, ok, err := s.conn.HearFirst(ctx, subjects, timeout, func(body json.RawMessage) bool { return bodyMatches(body, filter) }) + heard, ok, refused, err := s.conn.HearFirst(ctx, subjects, timeout, func(body json.RawMessage) bool { return bodyMatches(body, filter) }) if err != nil { return failure("mesh_wait could not wait: " + err.Error()) } + var answer map[string]any if !ok { - return answerText(map[string]any{"timed-out": true, "waited": timeout.String(), "subjects": names, - "read": "nothing matching was heard; read the state once from the verb that owns it, then wait again"}) + answer = map[string]any{"timed-out": true, "waited": timeout.String(), "subjects": names, + "read": "nothing matching was heard; read the state once from the verb that owns it, then wait again"} + } else { + answer = map[string]any{ + "heard": map[string]any{"subject": family[heard.Subject], "body": heard.Body, "at": heard.At.Format(time.RFC3339Nano)}, + "read": "an event only wakes you: read the state from the verb that owns it before acting (novox/hq ADR 0367 §4)", + } } - return answerText(map[string]any{ - "heard": map[string]any{"subject": family[heard.Subject], "body": heard.Body, "at": heard.At.Format(time.RFC3339Nano)}, - "read": "an event only wakes you: read the state from the verb that owns it before acting (novox/hq ADR 0367 §4)", - }) + // A subscription the bus refused while others stood is said (design 34 every-event §16). + if len(refused) > 0 { + answer["refused"] = refused + } + return answerText(answer) } // waitNames is the families a call names: a list of names, or one name.