Q7: the assigner reads the operator-channel seat's asks at its consumer's bind and asks again each held asking that lapsed or was cancelled (issue 1966, ADR 0668 §4)

An ask-ended the teller said before the assigner's consumer existed is not delivered to it (note 155).
At each bind the assigner calls the seat verb operator-channel.asks once, a kept call, for the ended
asks since its earliest held asking, and writes again each one that ended expired or cancelled, by
the ask-ended rules and their compare-and-set. A seat verb: no module named. One invoke added.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
2026-11-10 03:10:33 +01:00
parent 7bdc41c0d3
commit 3f2482716b
9 changed files with 229 additions and 4 deletions
@@ -11,6 +11,8 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents"
"git.novox.be/novox/mesh-sdk/go/process"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
"mesh-delivery-assigner/internal/assign"
"mesh-delivery-assigner/internal/buckets"
"mesh-delivery-assigner/internal/puts"
@@ -40,6 +42,76 @@ func hearAskEnded(ctx context.Context, js jetstream.JetStream, hub *process.Even
})
}
// channelAsksVerb is the operator-channel seat's verb asks, read at the assigner's consumer's bind (a seat verb: no module
// named).
const channelAsksVerb = "seat:" + oc.Seat + "." + oc.VerbAsks
// readEndedAtBind registers on the hub, before it starts, one read of the operator-channel seat's asks at each bind of the
// assigner's consumer: an ask-ended the teller said before the consumer existed is not delivered to it (issue 1966 note 155,
// ADR 0668 §4), so each held ask whose asking ended there expired or cancelled is written again as that event writes it
// (askAgain, with every rule and its compare-and-set). The read is a kept call (CallKept), made off the hub's start.
func readEndedAtBind(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, asksKV jetstream.KeyValue, bus caller, logf func(string, ...any)) {
if asksKV == nil || bus == nil {
return
}
hub.OnBound(func(created time.Time) {
go func() {
if err := endedAtBind(ctx, js, asksKV, bus, created, time.Now(), logf); err != nil {
logf("the operator channel's ended asks are not read at the bind of %s (%v): an ask that ended there before is asked again when the teller says it ended", created.Format(time.RFC3339), err)
}
}()
})
}
// endedAtBind is readEndedAtBind's one read: the held asks, the seat's verb, and each ended asking written again.
func endedAtBind(ctx context.Context, js jetstream.JetStream, asksKV jetstream.KeyValue, bus caller, created, now time.Time, logf func(string, ...any)) error {
keys, err := asksKV.Keys(ctx)
if errors.Is(err, jetstream.ErrNoKeysFound) {
return nil
}
if err != nil {
return err
}
held := map[string][]byte{}
for _, key := range keys {
if e, err := asksKV.Get(ctx, key); err == nil {
held[key] = e.Value()
}
}
query, ok := assign.ChannelAsksQuery(held)
if !ok {
return nil
}
answer, err := bus.CallKept(channelAsksVerb, query)
if err != nil {
return err
}
ended, cut, err := assign.EndedAtChannel(held, answer)
if err != nil {
return err
}
if cut != "" {
logf("%s at the bind: %s", channelAsksVerb, cut)
}
var putsKV jetstream.KeyValue
if js != nil {
if kv, err := js.KeyValue(ctx, buckets.Puts); err == nil {
putsKV = kv
}
}
for _, e := range ended {
body, err := tellerevents.EncodeEnded(e)
if err == nil {
err = askAgain(ctx, asksKV, putsKV, body, now, logf)
}
if err != nil {
return err
}
}
logf("%s at the bind of %s: %d of %d held asks ended at the channel without an answer", channelAsksVerb, created.Format(time.RFC3339), len(ended), len(held))
return nil
}
// askAgain writes the ask the teller said ended again under a new asking, by compare-and-set at the revision that holds the
// asking that ended (assign.AskedAgain), and says in the journal what it wrote or why it wrote nothing. Only a bus that
// could not be read or written is an error, so the event is heard again; a body that does not read is said and dropped.
@@ -1103,7 +1103,7 @@ func TestAHoldWrittenAgainAfterALiftAndARestartIsAskedAgain(t *testing.T) {
time.Sleep(1100 * time.Millisecond) // the next hold is written in another second
b.answerTo(t, akey, "keep", first.Ask.ID) // an answer to the first hold, which the later hold must not take
b.putJSON(t, buckets.Have, "ace.web", failed)
b.start(t)
_, stop = b.startStoppable(t)
again, err := decodeAsk(akey, b.get(t, buckets.Asks, akey))
if err != nil {
t.Fatal(err)
@@ -1111,6 +1111,34 @@ func TestAHoldWrittenAgainAfterALiftAndARestartIsAskedAgain(t *testing.T) {
if again.Ask.ID == first.Ask.ID {
t.Fatalf("the hold written again is asked under the first id %s", first.Ask.ID)
}
// Issue 1966 note 155, ADR 0668 §4: that asking lapses at the operator channel while no assigner runs, so the teller's
// ask-ended is said before the next one's consumer binds; at its bind the next reads the operator-channel seat's asks and
// writes the hold again under a new asking.
stop()
nc, err := nats.Connect(b.srv.ClientURL())
if err != nil {
t.Fatal(err)
}
defer nc.Close()
for _, subject := range []string{"mesh.seat.operator-channel.tool.asks", "mesh.seat.operator-channel.call.asks.*"} {
if _, err := nc.Subscribe(subject, func(m *nats.Msg) {
answer, _ := json.Marshal(map[string]any{"asks": []map[string]string{{"key": "mesh-delivery-teller." + again.Ask.ID, "state": "expired"}}, "matched": 1})
reply, _ := json.Marshal(map[string]any{"result": json.RawMessage(answer)})
_ = m.Respond(reply)
}); err != nil {
t.Fatal(err)
}
}
subscribed(t, nc)
b.start(t)
for deadline := time.Now().Add(15 * time.Second); ; time.Sleep(50 * time.Millisecond) {
if third, err := decodeAsk(akey, b.get(t, buckets.Asks, akey)); err == nil && third.Ask.ID != again.Ask.ID {
break
}
if time.Now().After(deadline) {
t.Fatalf("the hold whose asking %s lapsed before the assigner's bind was not asked again", again.Ask.ID)
}
}
b.putJSON(t, buckets.Have, "ace.web", healthy)
for deadline := time.Now().Add(15 * time.Second); b.hasKey(buckets.Wants, hkey); time.Sleep(100 * time.Millisecond) {
if time.Now().After(deadline) {
@@ -12,11 +12,11 @@ import (
"fmt"
"io"
"log"
"maps"
"mesh-delivery-assigner/internal/assignment"
"mesh-delivery-assigner/internal/buckets"
"os"
"os/signal"
"maps"
"slices"
"syscall"
"time"
@@ -483,6 +483,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
if err := hearAskEnded(ctx, js, hub, kvs[buckets.Asks], logger.Printf); err != nil {
logger.Printf("%s is not heard, so an ask that ends at the operator channel without an answer is not asked again until the assigner starts again: %v", askEndedSubject, err)
}
readEndedAtBind(ctx, js, hub, kvs[buckets.Asks], bus, logger.Printf)
hearReadsChanged(ctx, js, hub, reads, findings, gate.eventsLive, logger.Printf)
}
if bus != nil && kvs[buckets.Puts] != nil {
@@ -316,6 +316,7 @@ func (b *fakeBus) Call(key string, body any) (json.RawMessage, error) {
func (b *fakeBus) CallOn(node, key string, body any) (json.RawMessage, error) {
return b.Call(node+"/"+key, body)
}
func (b *fakeBus) CallKept(key string, body any) (json.RawMessage, error) { return b.Call(key, body) }
func (b *fakeBus) CallKeptOn(node, key string, body any) (json.RawMessage, error) {
return b.CallOn(node, key, body)
}
@@ -41,6 +41,8 @@ const PutEvent = "put"
type caller interface {
Call(key string, body any) (json.RawMessage, error)
CallOn(node, key string, body any) (json.RawMessage, error)
// CallKept is Call for a kept call, safe to call again (process.Conn.CallKept; ADR 0668).
CallKept(key string, body any) (json.RawMessage, error)
// CallKeptOn is CallOn for a kept call, safe to call again (process.Conn.CallKeptOn; ADR 0668).
CallKeptOn(node, key string, body any) (json.RawMessage, error)
Emit(eventType string, body any) error
@@ -28,6 +28,7 @@ func TestTheHoldAskIsValid(t *testing.T) {
t.Run("hold", theHoldAskIsValid)
t.Run("restore", theRestoreAskIsValid)
t.Run("ended at the channel", anAskEndedAtTheChannelIsWrittenAgain)
t.Run("ended at the channel, read at bind", anAskEndedAtTheChannelIsReadAtBind)
}
// Issue 1966, note 151: an ask the teller says ended at the operator channel without an answer (its event ask-ended) is
@@ -122,6 +123,50 @@ func anAskEndedAtTheChannelIsWrittenAgain(t *testing.T) {
}
}
// Issue 1966 note 155, ADR 0668 §4: an ended ask the teller said before the assigner's consumer existed is not delivered to
// it, so at bind the assigner reads the operator-channel seat's asks: the query reaches back to the earliest held asking,
// and of the answer only the held askings that lapsed or were cancelled are taken, as the teller's event says them; an
// answered or another asker's ask, and one the key no longer holds, are not.
func anAskEndedAtTheChannelIsReadAtBind(t *testing.T) {
asked := time.Date(2026, 11, 9, 22, 0, 0, 0, time.UTC)
holdKey, _ := asks.HoldAskKey("nats", fNew)
restoreKey, _ := asks.RestoreAskKey("novox", "nats", fNew)
backupKey, _ := asks.BackupAskKey("novox", "nats", fNew)
hold := HoldAsk(holdKey, "nats", fNew, "novox", "w", AskingSeconds(asked), asked)
restore := RestoreAsk(restoreKey, "novox", "nats", fNew, "w", AskingSeconds(asked.Add(time.Hour)), asked.Add(time.Hour))
backup := BackupAsk(backupKey, "novox", "nats", fNew, "sha256:"+strings.Repeat("e", 64), "w", 0, asked.Add(-time.Hour)) // no asking
held := map[string][]byte{}
for key, a := range map[string]asks.Ask{holdKey: hold, restoreKey: restore, backupKey: backup} {
v, err := EncodeAsk(key, a)
if err != nil {
t.Fatal(err)
}
held[key] = v
}
if _, ok := ChannelAsksQuery(nil); ok {
t.Error("with no ask held the channel is asked")
}
q, ok := ChannelAsksQuery(held)
if since, _ := q["since"].(string); !ok || q["state"] != "ended" || since != asked.Add(-time.Hour-time.Minute).Format(time.RFC3339) {
t.Errorf("the query is %v, want the ended asks since %s", q, asked.Add(-time.Hour-time.Minute).Format(time.RFC3339))
}
at := func(key string, a asks.Ask, state string) string {
return fmt.Sprintf(`{"key":"t.%s","state":%q}`, mustAsk(t, a).ID, state)
}
answer := `{"asks":[` + strings.Join([]string{at(holdKey, hold, occ.OutcomeExpired), at(restoreKey, restore, occ.OutcomeChosen),
at(backupKey, backup, occ.OutcomeCancelled), `{"key":"t.hold-gone","state":"expired"}`, `{"key":"open","state":"open"}`}, ",") + `],"cut":"c"}`
ended, cut, err := EndedAtChannel(held, []byte(answer))
want := []tellerevents.Ended{{Key: backupKey, Ask: mustAsk(t, backup).ID, Outcome: occ.OutcomeCancelled},
{Key: holdKey, Ask: mustAsk(t, hold).ID, Outcome: occ.OutcomeExpired}}
slices.SortFunc(want, func(a, b tellerevents.Ended) int { return strings.Compare(a.Key, b.Key) })
if err != nil || cut != "c" || !slices.Equal(ended, want) {
t.Errorf("EndedAtChannel = %+v, %q, %v; want %+v and the answer's cut", ended, cut, err, want)
}
if _, _, err := EndedAtChannel(held, []byte(`[1]`)); err == nil {
t.Error("an answer that does not read is taken")
}
}
func ptr(a asks.Ask) *asks.Ask { return &a }
func theHoldAskIsValid(t *testing.T) {
@@ -1,10 +1,14 @@
package assign
import (
"encoding/json"
"fmt"
"slices"
"strings"
"time"
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
@@ -71,3 +75,73 @@ func AskedAgain(e tellerevents.Ended, current []byte, put *puts.Put, now time.Ti
}
return value, ""
}
// ChannelAsksBound is the most bytes of asks the assigner asks the operator-channel seat's verb asks for at its consumer's
// bind: every ended asking of its held asks fits.
const ChannelAsksBound = 1 << 20
// ChannelAsksQuery is the arguments of the operator-channel seat's verb asks that answer, at the assigner's consumer's bind,
// every asking of its held asks (key to value, as the asks state holds them) that ended at the channel: the ended asks
// opened since the earliest of them could have been opened, less a minute (issue 1966 note 155, ADR 0668 §4). ok is false
// when no held ask reads, and then nothing is asked.
func ChannelAsksQuery(held map[string][]byte) (query map[string]any, ok bool) {
var since time.Time
for key, value := range held {
a, ask, err := DecodeAsk(key, value)
if err != nil {
continue
}
opened := ExpiresAt(ask).Add(-AskLasts)
if a.Asking >= 2 {
if at := time.Unix(int64(a.Asking), 0); at.Before(opened) {
opened = at
}
}
if !ok || opened.Before(since) {
since, ok = opened, true
}
}
if !ok {
return nil, false
}
return map[string]any{"state": "ended", "since": since.Add(-time.Minute).UTC().Format(time.RFC3339), "bound": ChannelAsksBound}, true
}
// channelAsks is what the assigner reads of the verb asks' answer: each ask's key ("<asker>.<id>", the seat's state asks)
// and the state it ended with.
type channelAsks struct {
Asks []struct {
Key string `json:"key"`
State string `json:"state"`
} `json:"asks"`
Cut string `json:"cut"`
}
// EndedAtChannel is, of the operator-channel seat's answer of its verb asks, each held ask whose asking ended there
// expired or cancelled, as the teller says it in its event ask-ended, so the assigner writes it again by the rules of that
// event (AskedAgain), with its compare-and-set: an ended ask said before the assigner's consumer existed is not delivered
// to it (issue 1966 note 155, ADR 0668 §4). cut is the answer's own cut, said by the caller; an answer that does not read
// is the error.
func EndedAtChannel(held map[string][]byte, answer []byte) (ended []tellerevents.Ended, cut string, err error) {
var got channelAsks
if err := json.Unmarshal(answer, &got); err != nil {
return nil, "", fmt.Errorf("assign: the operator channel's asks do not read: %w", err)
}
byID := map[string]string{}
for key, value := range held {
if _, ask, err := DecodeAsk(key, value); err == nil {
byID[ask.ID] = key
}
}
for _, v := range got.Asks {
if v.State != occ.OutcomeExpired && v.State != occ.OutcomeCancelled {
continue
}
_, id, ok := oc.ParseAsksByAskerIDKey(v.Key)
if key, held := byID[id]; ok && held {
ended = append(ended, tellerevents.Ended{Key: key, Ask: id, Outcome: v.State})
}
}
slices.SortFunc(ended, func(a, b tellerevents.Ended) int { return strings.Compare(a.Key, b.Key) })
return ended, got.Cut, nil
}
@@ -130,7 +130,8 @@ func TestManifest(t *testing.T) {
"seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list
"seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat alone
"seat:node-backup.backed-up", "seat:node-backup.now",
"seat:node-engine.manifest-reads"}; !slices.Equal(slices.Sorted(slices.Values(m.Invokes)), want) { // ADR 0618 §3
"seat:node-engine.manifest-reads", // ADR 0618 §3
"seat:operator-channel.asks"}; !slices.Equal(slices.Sorted(slices.Values(m.Invokes)), want) { // ADR 0668 §4: the ended asks at bind
t.Errorf("invokes = %v, want %v", m.Invokes, want)
}
if want := []string{"assignments-copied", "put", "views.asks", "views.copies", "views.differences", "views.findings", "views.assignments", "views.seat-holders", "views.wants"}; !slices.Equal(m.Emits, want) {
+2 -1
View File
@@ -57,7 +57,8 @@
"seat:mesh-hand-acts.warranted",
"seat:node-backup.backed-up",
"seat:node-backup.now",
"seat:node-engine.manifest-reads"
"seat:node-engine.manifest-reads",
"seat:operator-channel.asks"
],
"emits": [
"assignments-copied",