Subtask A: the assigner hears the teller's ask-ended and writes the ask again under a new asking (issue 1966)

An ask that ended at the operator channel without an answer while its key stood was never
written again, so nothing asked it (note 151: the nats yes on novox, three backup asks). The
assigner now consumes mesh-delivery-teller.ask-ended on its durable consumer and writes the key
again by compare-and-set under AskingSeconds (above the old asking), an id of its own and a new
expiry (AskLasts), one row per kind: hold, restore, backup and a record ask of the rules are
written again; the authorising ask of a proposal is not (ADR 0567 §5, §11: its lapse ends the
proposal, again asks it again), nor a record ask a put asked (ADR 0546: its lapse refuses the
put), nor an asking the key no longer holds.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
2026-11-10 01:10:35 +01:00
parent b39b0771a0
commit 46badf078e
8 changed files with 282 additions and 7 deletions
@@ -0,0 +1,94 @@
package main
import (
"context"
"errors"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents"
"git.novox.be/novox/mesh-sdk/go/process"
"mesh-delivery-assigner/internal/assign"
"mesh-delivery-assigner/internal/buckets"
"mesh-delivery-assigner/internal/puts"
)
// askEndedSubject is the teller's event ask-ended, as the manifest's consumes names it (`mesh-delivery-teller.ask-ended`).
var askEndedSubject = process.Account{Module: buckets.TellerModule}.EventSubject(tellerevents.AskEnded)
// hearAskEnded hears the teller's ask-ended on the assigner's one hub of its durable consumer on EVENTS (registered before
// the hub starts): an ask of the assigner's that ended at the operator channel without an answer while its key stands is
// written again under a new asking, so the teller asks it again (issue 1966, note 151). asksKV is the assigner's asks; the
// puts are opened at each event, since a put's record ask is not written again (ADR 0546).
func hearAskEnded(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, asksKV jetstream.KeyValue, logf func(string, ...any)) error {
return hub.Handle(askEndedSubject, func(m jetstream.Msg) {
var putsKV jetstream.KeyValue
if kv, err := js.KeyValue(ctx, buckets.Puts); err == nil {
putsKV = kv
}
if err := askAgain(ctx, asksKV, putsKV, m.Data(), time.Now(), logf); err != nil {
logf("%s: %v; it is heard again", m.Subject(), err)
_ = m.Nak()
return
}
if err := m.Ack(); err != nil {
logf("acknowledging %s: %v", m.Subject(), err)
}
})
}
// 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.
func askAgain(ctx context.Context, asksKV, putsKV jetstream.KeyValue, data []byte, now time.Time, logf func(string, ...any)) error {
e, err := tellerevents.DecodeEnded(data)
if err != nil {
logf("the teller's %s holds what is not an ended ask, and is dropped: %v", tellerevents.AskEnded, err)
return nil
}
if asksKV == nil {
return errors.New("the assigner's asks are not open")
}
var current []byte
var rev uint64
entry, err := asksKV.Get(ctx, e.Key)
switch {
case errors.Is(err, jetstream.ErrKeyNotFound):
case err != nil:
return err
default:
current, rev = entry.Value(), entry.Revision()
}
var put *puts.Put
if k, err := asks.ParseAnyAskKey(e.Key); err == nil && k.Kind == asks.AskKindRecord && putsKV != nil {
if pkey, err := puts.Key(k.Node, k.Module); err == nil {
pe, err := putsKV.Get(ctx, pkey)
switch {
case errors.Is(err, jetstream.ErrKeyNotFound):
case err != nil:
return err
default:
if p, err := puts.Decode(pe.Value()); err == nil {
put = &p
}
}
}
}
value, why := assign.AskedAgain(e, current, put, now)
if why != "" {
logf("%s: the ask %s ended (%s) at the operator channel without an answer, and is not written again: %s", e.Key, e.Ask, e.Outcome, why)
return nil
}
if _, err := asksKV.Update(ctx, e.Key, value, rev); err != nil {
if errors.Is(err, jetstream.ErrKeyExists) {
logf("%s: the ask %s ended (%s) without an answer, and the key changed since: it is not written again", e.Key, e.Ask, e.Outcome)
return nil
}
return err
}
logf("%s: the ask %s ended (%s) at the operator channel without an answer, and is written again under a new asking, which the teller asks", e.Key, e.Ask, e.Outcome)
return nil
}
@@ -477,6 +477,11 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
// Heard on the assigner's durable consumer on EVENTS, which the grants part makes from the manifest's consumes; where it
// is not there yet, a node is asked again only when it comes back online.
if js != nil {
// The teller's ask-ended (issue 1966): an ask ended at the channel without an answer is written again; registered
// before hearReadsChanged starts the hub.
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)
}
hearReadsChanged(ctx, js, hub, reads, findings, gate.eventsLive, logger.Printf)
}
if bus != nil && kvs[buckets.Puts] != nil {
+4
View File
@@ -74,8 +74,12 @@ require git.novox.be/novox/mesh-contracts/identity-and-access/go/seats/meshnodee
require git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks v0.0.0-00010101000000-000000000000
require git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents v0.0.0-00010101000000-000000000000
replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks => ./asks
replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents => ../mesh-delivery-teller/tellerevents
require git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate v0.0.0-00010101000000-000000000000
replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate => ../mesh-delivery-view/viewstate
@@ -32,7 +32,7 @@ func HoldAsk(key, module, fp, node, words string, asking int, now time.Time) ask
{ID: "release", Label: "Release it", Does: "the version is released and the wants move to it by the rules again", Level: occ.Acknowledge},
},
Who: occ.Operator,
Expires: occ.FormatTime(now.Add(7 * 24 * time.Hour)),
Expires: occ.FormatTime(now.Add(AskLasts)),
OnExpiry: "the version stays held",
About: key,
})
@@ -60,7 +60,7 @@ func RestoreAsk(key, node, module, fp, words string, asking int, now time.Time)
{ID: OptionLeaveStopped, Label: "Leave it stopped", Does: "the module stays as it is on the node and the data is kept", Level: occ.Acknowledge},
},
Who: occ.Operator,
Expires: occ.FormatTime(now.Add(7 * 24 * time.Hour)),
Expires: occ.FormatTime(now.Add(AskLasts)),
OnExpiry: "the module stays as it is on the node",
About: key,
})
@@ -181,7 +181,7 @@ func RecordAsk(key, node, module, fp string, need RecordNeed, asking int, now ti
{ID: OptionNo, Label: "No, not now", Does: "the node keeps what it runs", Level: occ.Acknowledge},
},
Who: occ.Operator,
Expires: occ.FormatTime(now.Add(7 * 24 * time.Hour)),
Expires: occ.FormatTime(now.Add(AskLasts)),
OnExpiry: "the node keeps what it runs",
About: key,
})
@@ -236,7 +236,7 @@ func BackupAsk(key, node, module, fp, digest, words string, asking int, now time
{ID: OptionBackupSkip, Label: "Go on without a backup", Does: "the node applies the new version without a backup", Level: occ.Acknowledge},
},
Who: occ.Operator,
Expires: occ.FormatTime(now.Add(7 * 24 * time.Hour)),
Expires: occ.FormatTime(now.Add(AskLasts)),
OnExpiry: "the module keeps running what it has",
About: key,
})
@@ -2,6 +2,7 @@ package assign
import (
"fmt"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents"
"mesh-delivery-assigner/internal/assignment"
"mesh-delivery-assigner/internal/puts"
"slices"
@@ -26,8 +27,103 @@ import (
func TestTheHoldAskIsValid(t *testing.T) {
t.Run("hold", theHoldAskIsValid)
t.Run("restore", theRestoreAskIsValid)
t.Run("ended at the channel", anAskEndedAtTheChannelIsWrittenAgain)
}
// Issue 1966, note 151: an ask the teller says ended at the operator channel without an answer (its event ask-ended) is
// written again under a new asking, an id of its own and a new expiry, one row per kind: the rules' asks are; the
// authorising ask of a proposal (ADR 0567 §5, §11) and a put's record ask (ADR 0546) are not, nor an asking the key no longer
// holds, an end asking again would not change, or a body that does not read.
func anAskEndedAtTheChannelIsWrittenAgain(t *testing.T) {
asked := time.Date(2026, 11, 9, 22, 0, 0, 0, time.UTC)
now := asked.Add(8 * 24 * time.Hour)
digest := "sha256:" + strings.Repeat("e", 64)
holdKey, _ := asks.HoldAskKey("nats", fNew)
restoreKey, _ := asks.RestoreAskKey("novox", "nats", fNew)
backupKey, _ := asks.BackupAskKey("novox", "nats", fNew)
recordKey, _ := asks.RecordAskKey("novox", "nats", digest)
assignmentsKey, _ := asks.AssignmentsAskKey("p-20261028t183000-1a2b3c")
need := RecordNeed{Node: "novox", Digest: digest, Judgement: JudgedMovesData}
record := RecordAsk(recordKey, "novox", "nats", fNew, need, 0, asked) // written before #195: no asking
put := puts.Put{FP: fNew, Why: "w", At: asked, Step: puts.Person, Ask: recordKey}
byPut := RecordAsk(recordKey, "novox", "nats", fNew, need, PutAsking(put), asked)
other := puts.Put{FP: fNew, Why: "w", At: asked.Add(time.Hour), Step: puts.Person, Ask: recordKey}
assignments := assignerAsk(AskingSeconds(asked), oc.Ask{ID: meshdeliveryclient.AskingID(assignmentsKey, AskingSeconds(asked)), Headline: "h",
Explanation: "e", Who: occ.Operator, Expires: occ.FormatTime(asked.Add(occ.ApproveLasts)), OnExpiry: "o",
Options: []oc.Option{{ID: OptionYes, Label: "Yes", Does: "d", Level: occ.Approve}}})
for _, c := range []struct {
name string
key string
a *asks.Ask // the ask the key holds; nil for none
ended string // the id of the asking that ended; "" for the one the key holds
outcome string
put *puts.Put
why string // "" when it is written again
}{
{name: "hold", key: holdKey, a: ptr(HoldAsk(holdKey, "nats", fNew, "novox", "w", AskingSeconds(asked), asked)), outcome: occ.OutcomeExpired},
{name: "restore", key: restoreKey, a: ptr(RestoreAsk(restoreKey, "novox", "nats", fNew, "w", AskingSeconds(asked), asked)), outcome: occ.OutcomeCancelled},
{name: "backup", key: backupKey, a: ptr(BackupAsk(backupKey, "novox", "nats", fNew, digest, "w", 0, asked)), outcome: occ.OutcomeCancelled},
{name: "record", key: recordKey, a: &record, outcome: occ.OutcomeCancelled},
{name: "record of another put", key: recordKey, a: &record, outcome: occ.OutcomeExpired, put: &other},
{name: "record a put asked", key: recordKey, a: &byPut, outcome: occ.OutcomeExpired, put: &put, why: "put"},
{name: "assignments", key: assignmentsKey, a: &assignments, outcome: occ.OutcomeExpired, why: "ADR 0567"},
{name: "another asking stands", key: holdKey, a: ptr(HoldAsk(holdKey, "nats", fNew, "novox", "w", AskingSeconds(now), now)),
ended: meshdeliveryclient.AskingID(holdKey, AskingSeconds(asked)), outcome: occ.OutcomeExpired, why: "holds the asking"},
{name: "taken away", key: holdKey, ended: "hold-x", outcome: occ.OutcomeExpired, why: "taken away"},
{name: "refused", key: holdKey, a: ptr(HoldAsk(holdKey, "nats", fNew, "novox", "w", 0, asked)), outcome: occ.OutcomeRefused, why: "refused"},
{name: "replaced", key: holdKey, a: ptr(HoldAsk(holdKey, "nats", fNew, "novox", "w", 0, asked)), outcome: occ.OutcomeReplaced, why: "replaced"},
} {
t.Run(c.name, func(t *testing.T) {
var current []byte
ended := c.ended
if c.a != nil {
var err error
if current, err = EncodeAsk(c.key, *c.a); err != nil {
t.Fatal(err)
}
if ended == "" {
ended = mustAsk(t, *c.a).ID
}
}
body, err := tellerevents.EncodeEnded(tellerevents.Ended{Key: c.key, Ask: ended, Outcome: c.outcome})
if err != nil {
t.Fatal(err)
}
e, err := tellerevents.DecodeEnded(body)
if err != nil {
t.Fatal(err)
}
value, why := AskedAgain(e, current, c.put, now)
if c.why != "" {
if value != nil || !strings.Contains(why, c.why) {
t.Fatalf("AskedAgain = %s, %q; want nothing written, %q", value, why, c.why)
}
return
}
if why != "" {
t.Fatalf("AskedAgain says %q, want the ask written again", why)
}
a, ask, err := DecodeAsk(c.key, value)
if err != nil {
t.Fatal(err)
}
if a.Asking <= c.a.Asking || ask.ID != meshdeliveryclient.AskingID(c.key, a.Asking) || ask.ID == ended {
t.Errorf("asking %d under %s, want a new asking above %d under an id of its own, not %s", a.Asking, ask.ID, c.a.Asking, ended)
}
if err := occ.Check(ask, now); err != nil || !ExpiresAt(ask).Equal(now.Add(AskLasts).Truncate(time.Second)) {
t.Errorf("the router refuses the ask written again (%v), or it lapses at %s, want %s", err, ExpiresAt(ask), now.Add(AskLasts))
}
})
}
for _, raw := range []string{`{"v":1,"key":"k","ask":"a"}`, `{"v":1,"key":"k","ask":"a","outcome":"expired","more":1}`} {
if _, err := tellerevents.DecodeEnded([]byte(raw)); err == nil {
t.Errorf("DecodeEnded(%s) takes a body that is not an ended ask", raw)
}
}
}
func ptr(a asks.Ask) *asks.Ask { return &a }
func theHoldAskIsValid(t *testing.T) {
key, err := asks.HoldAskKey("mail", fNew)
if err != nil {
@@ -0,0 +1,73 @@
package assign
import (
"fmt"
"time"
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents"
"mesh-delivery-assigner/internal/puts"
)
// AskLasts is how long an ask of the assigner's rules stands at the operator channel before it lapses unanswered: the hold,
// restore, record and backup asks.
const AskLasts = 7 * 24 * time.Hour
// writtenAgainOnEnd is, per kind of the assigner's asks, whether an asking that ended at the channel without an answer while
// its key stands is written again under a new asking (issue 1966, note 151), or why not. A rule's ask waits for a person
// as long as its key stands, so it is asked again; the authorising ask of a proposal ends the proposal when it lapses
// (ADR 0567 §5) and is asked again by the seat's verb `again` (§11).
var writtenAgainOnEnd = map[string]string{
asks.AskKindHold: "",
asks.AskKindRestore: "",
asks.AskKindBackup: "",
asks.AskKindRecord: "",
asks.AskKindAssignments: "a proposal's authorising ask that lapsed ends the proposal (ADR 0567 §5); the verb again asks it again (§11)",
}
// AskedAgain is the ask at key written again after the teller said its asking ended at the channel without an answer (the
// teller's event ask-ended): the value to write by compare-and-set over current, or why nothing is written. Nothing is written
// for an outcome other than a lapse or a cancellation, for a key that holds another asking now (an answer to it, or a newer
// asking, came since), for a kind its own rule ends (writtenAgainOnEnd), and for a record ask a put asked (put, nil for none:
// the put at the record ask's node and module), since its put is refused when it lapses (ADR 0546). The new asking is
// AskingSeconds(now), above the old one, under an id of its own, standing AskLasts from now.
func AskedAgain(e tellerevents.Ended, current []byte, put *puts.Put, now time.Time) ([]byte, string) {
if e.Outcome != occ.OutcomeExpired && e.Outcome != occ.OutcomeCancelled {
return nil, fmt.Sprintf("it ended %s, which asking again would not change", e.Outcome)
}
k, err := asks.ParseAnyAskKey(e.Key)
if err != nil {
return nil, err.Error()
}
why, known := writtenAgainOnEnd[k.Kind]
if !known {
return nil, "the assigner asks no " + k.Kind + " ask"
}
if why != "" {
return nil, why
}
if current == nil {
return nil, "the key is taken away: nothing is asked"
}
a, ask, err := DecodeAsk(e.Key, current)
if err != nil {
return nil, err.Error()
}
if ask.ID != e.Ask {
return nil, "the key holds the asking " + ask.ID + " now, not the one that ended"
}
if k.Kind == asks.AskKindRecord && put != nil && put.Ask == e.Key && PutAsking(*put) == a.Asking {
return nil, "a put asked it, and its put is refused when it lapses (ADR 0546)"
}
asking := max(AskingSeconds(now), a.Asking+1)
ask.ID, ask.Expires = meshdeliveryclient.AskingID(e.Key, asking), occ.FormatTime(now.Add(AskLasts))
value, err := EncodeAsk(e.Key, assignerAsk(asking, ask))
if err != nil {
return nil, err.Error()
}
return value, ""
}
@@ -3,6 +3,7 @@ package manifest
import (
"git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents"
"os"
"reflect"
"slices"
@@ -120,9 +121,10 @@ func TestManifest(t *testing.T) {
t.Errorf("emits lists %s, an event the seat says", e)
}
}
// ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events.
if !slices.Equal(m.Consumes, []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", "node-engine.manifest-reads-changed"}) {
t.Errorf("consumes = %v, want the indexer's views of graph, manifests and modules and the engines' reads changed", m.Consumes)
// ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events;
// issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears.
if !slices.Equal(m.Consumes, []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "node-engine.manifest-reads-changed"}) {
t.Errorf("consumes = %v, want the indexer's views of graph, manifests and modules, the teller's ask-ended and the engines' reads changed", m.Consumes)
}
if want := []string{"mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules",
"seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list
@@ -44,6 +44,7 @@
"mesh-delivery-indexer.views.graph",
"mesh-delivery-indexer.views.manifests",
"mesh-delivery-indexer.views.modules",
"mesh-delivery-teller.ask-ended",
"node-engine.manifest-reads-changed"
],
"invokes": [