diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/askended.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/askended.go index e121b991..893232ee 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/askended.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/askended.go @@ -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. diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index b4267224..b6e8c91e 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -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) { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 48ced1a4..1f51d4bc 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -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 { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index ada4fe06..f760fc08 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -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) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go index 18540885..929ecbdb 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go @@ -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 diff --git a/modules/mesh-delivery-assigner/internal/assign/asks_test.go b/modules/mesh-delivery-assigner/internal/assign/asks_test.go index 8ac49816..d550f1fa 100644 --- a/modules/mesh-delivery-assigner/internal/assign/asks_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/asks_test.go @@ -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) { diff --git a/modules/mesh-delivery-assigner/internal/assign/ended.go b/modules/mesh-delivery-assigner/internal/assign/ended.go index 107db2b2..8b5e4e7f 100644 --- a/modules/mesh-delivery-assigner/internal/assign/ended.go +++ b/modules/mesh-delivery-assigner/internal/assign/ended.go @@ -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 (".", 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 +} diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 4438804c..7c0a697c 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -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) { diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 63ae125a..bb459b24 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -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",