Merge pull request 'An assigner ask that ends at the channel without an answer is asked again under a new asking (issue 1966)' (#198) from fix/1966-ask-ended into main
This commit was merged in pull request #198.
This commit is contained in:
@@ -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
|
// 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.
|
// is not there yet, a node is asked again only when it comes back online.
|
||||||
if js != nil {
|
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)
|
hearReadsChanged(ctx, js, hub, reads, findings, gate.eventsLive, logger.Printf)
|
||||||
}
|
}
|
||||||
if bus != nil && kvs[buckets.Puts] != nil {
|
if bus != nil && kvs[buckets.Puts] != nil {
|
||||||
|
|||||||
@@ -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-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-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
|
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
|
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},
|
{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,
|
Who: occ.Operator,
|
||||||
Expires: occ.FormatTime(now.Add(7 * 24 * time.Hour)),
|
Expires: occ.FormatTime(now.Add(AskLasts)),
|
||||||
OnExpiry: "the version stays held",
|
OnExpiry: "the version stays held",
|
||||||
About: key,
|
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},
|
{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,
|
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",
|
OnExpiry: "the module stays as it is on the node",
|
||||||
About: key,
|
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},
|
{ID: OptionNo, Label: "No, not now", Does: "the node keeps what it runs", Level: occ.Acknowledge},
|
||||||
},
|
},
|
||||||
Who: occ.Operator,
|
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",
|
OnExpiry: "the node keeps what it runs",
|
||||||
About: key,
|
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},
|
{ID: OptionBackupSkip, Label: "Go on without a backup", Does: "the node applies the new version without a backup", Level: occ.Acknowledge},
|
||||||
},
|
},
|
||||||
Who: occ.Operator,
|
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",
|
OnExpiry: "the module keeps running what it has",
|
||||||
About: key,
|
About: key,
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package assign
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents"
|
||||||
"mesh-delivery-assigner/internal/assignment"
|
"mesh-delivery-assigner/internal/assignment"
|
||||||
"mesh-delivery-assigner/internal/puts"
|
"mesh-delivery-assigner/internal/puts"
|
||||||
"slices"
|
"slices"
|
||||||
@@ -26,8 +27,103 @@ import (
|
|||||||
func TestTheHoldAskIsValid(t *testing.T) {
|
func TestTheHoldAskIsValid(t *testing.T) {
|
||||||
t.Run("hold", theHoldAskIsValid)
|
t.Run("hold", theHoldAskIsValid)
|
||||||
t.Run("restore", theRestoreAskIsValid)
|
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) {
|
func theHoldAskIsValid(t *testing.T) {
|
||||||
key, err := asks.HoldAskKey("mail", fNew)
|
key, err := asks.HoldAskKey("mail", fNew)
|
||||||
if err != nil {
|
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 (
|
import (
|
||||||
"git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
|
"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/contracts/go/seats/meshwants"
|
||||||
|
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents"
|
||||||
"os"
|
"os"
|
||||||
"reflect"
|
"reflect"
|
||||||
"slices"
|
"slices"
|
||||||
@@ -120,9 +121,10 @@ func TestManifest(t *testing.T) {
|
|||||||
t.Errorf("emits lists %s, an event the seat says", e)
|
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.
|
// 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"}) {
|
// issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears.
|
||||||
t.Errorf("consumes = %v, want the indexer's views of graph, manifests and modules and the engines' reads changed", m.Consumes)
|
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",
|
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
|
"seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list
|
||||||
|
|||||||
@@ -44,6 +44,7 @@
|
|||||||
"mesh-delivery-indexer.views.graph",
|
"mesh-delivery-indexer.views.graph",
|
||||||
"mesh-delivery-indexer.views.manifests",
|
"mesh-delivery-indexer.views.manifests",
|
||||||
"mesh-delivery-indexer.views.modules",
|
"mesh-delivery-indexer.views.modules",
|
||||||
|
"mesh-delivery-teller.ask-ended",
|
||||||
"node-engine.manifest-reads-changed"
|
"node-engine.manifest-reads-changed"
|
||||||
],
|
],
|
||||||
"invokes": [
|
"invokes": [
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ import (
|
|||||||
|
|
||||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
|
"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-assigner/asks"
|
||||||
|
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents"
|
||||||
)
|
)
|
||||||
|
|
||||||
// The assigner's asks reach the operator (issue 1906). The assigner writes its asks (the hold of a fingerprint, the restore
|
// The assigner's asks reach the operator (issue 1906). The assigner writes its asks (the hold of a fingerprint, the restore
|
||||||
@@ -163,15 +164,21 @@ func (t *Teller) raiseAssignerAsk(key string) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
if k != nil {
|
if k != nil {
|
||||||
|
byHand := grec != nil && prior.ID == id && prior.Hand
|
||||||
if grec == nil || prior.ID != id {
|
if grec == nil || prior.ID != id {
|
||||||
_ = t.note(id, key, 0, grec)
|
_ = t.note(id, key, 0, grec)
|
||||||
}
|
}
|
||||||
if open(k) {
|
switch {
|
||||||
|
case open(k):
|
||||||
t.catchUp(id)
|
t.catchUp(id)
|
||||||
} else {
|
case endedUnanswered(k) && !byHand:
|
||||||
// Said, never silent (issue 1966: a record ask taken back when its key went was written again under the same id on
|
// The key stands under an asking that ended at the channel without an answer, heard now or at the teller's start
|
||||||
// 2026-11-09 and nothing said why the operator was never asked): the assigner numbers each writing (asks.Ask.Asking).
|
// (issue 1966: the nats yes on novox and three backup asks stood so after #195, and nothing asked them again): the
|
||||||
t.logf("%s: the ask %s ended (answered, lapsed or taken back) and the assigner wrote the key again under the same id: not asked again, the router would drop it; a writing under a new asking is asked", key, id)
|
// assigner is told, and writes the key again under a new asking where its kind allows it.
|
||||||
|
t.sayAssignerAskEnded(key, id, k.Ended)
|
||||||
|
default:
|
||||||
|
// Said, never silent (issue 1966): an answered asking, or one a person ended by hand, is not asked again.
|
||||||
|
t.logf("%s: the ask %s ended (%s) and the assigner's key still stands under it: not asked again, the router would drop it; a writing under a new asking is asked", key, id, endedWords(k, byHand))
|
||||||
}
|
}
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -238,3 +245,42 @@ func (t *Teller) answerAssigner(key string, o oc.Option, w oc.Warrant) error {
|
|||||||
}
|
}
|
||||||
return t.writeAnswer(key, w.Ask, body, occ.Says(w))
|
return t.writeAnswer(key, w.Ask, body, occ.Says(w))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// endedUnanswered says whether an asking ended at the channel without an answer the assigner could ask again for: it lapsed,
|
||||||
|
// or it was cancelled (at the router, or by the teller when its key was taken away). A refusal would be refused again, and a
|
||||||
|
// replaced asking has a newer one standing.
|
||||||
|
func endedUnanswered(k *occ.Kept) bool {
|
||||||
|
return k != nil && !k.Acted && (k.Ended == occ.OutcomeExpired || k.Ended == occ.OutcomeCancelled)
|
||||||
|
}
|
||||||
|
|
||||||
|
// endedWords says how an ended asking ended, for the journal.
|
||||||
|
func endedWords(k *occ.Kept, byHand bool) string {
|
||||||
|
switch {
|
||||||
|
case byHand:
|
||||||
|
return "ended by hand"
|
||||||
|
case k.Acted:
|
||||||
|
return "answered"
|
||||||
|
}
|
||||||
|
return k.Ended
|
||||||
|
}
|
||||||
|
|
||||||
|
// sayAssignerAskEnded says the teller's event ask-ended for the assigner's ask at key whose asking id ended at the channel
|
||||||
|
// with outcome, once per asking (the bus drops a second message under its id): the assigner writes the key again under a new
|
||||||
|
// asking (issue 1966). An outcome the assigner could not ask again for is not said.
|
||||||
|
func (t *Teller) sayAssignerAskEnded(key, id, outcome string) {
|
||||||
|
if outcome != occ.OutcomeExpired && outcome != occ.OutcomeCancelled {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
body, err := tellerevents.EncodeEnded(tellerevents.Ended{Key: key, Ask: id, Outcome: outcome})
|
||||||
|
if err == nil && t.Emit == nil {
|
||||||
|
err = fmt.Errorf("the teller says no events here")
|
||||||
|
}
|
||||||
|
if err == nil {
|
||||||
|
err = t.Emit(tellerevents.AskEnded, body, tellerevents.AskEnded+"."+id)
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
t.logf("%s: the ask %s ended (%s) without an answer, and the assigner could not be told (%v): it is told again at the teller's next start", key, id, outcome, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
t.logf("%s: the ask %s ended (%s) without an answer: the assigner is told (%s), and writes the key again under a new asking where its kind allows it", key, id, outcome, tellerevents.AskEnded)
|
||||||
|
}
|
||||||
|
|||||||
@@ -1,9 +1,12 @@
|
|||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"slices"
|
||||||
|
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
|
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"
|
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
||||||
|
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
@@ -212,46 +215,91 @@ func TestABurstOfHoldsLeavesRoomForAMergeAsk(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Issue 1966 (nats on novox): an assigner ask taken back when its key went is asked again when the assigner writes the key
|
// Issue 1966 (nats on novox): an assigner ask taken back when its key went is asked again when the assigner writes the key
|
||||||
// again under a new asking, whatever its kind; written again under the id the router kept it is not (the router would drop it)
|
// again under a new asking, whatever its kind; written again under the id the router kept it is not (the router would drop it).
|
||||||
// and the teller says so, never silent.
|
// An asking that ended at the channel without an answer while its key stands (taken back and written again under its id, or
|
||||||
|
// lapsed at the router) is said to the assigner as the teller's event ask-ended, so the assigner writes it again under a new
|
||||||
|
// asking (note 151); one answered, refused, replaced or ended by hand is not.
|
||||||
func anAssignerAskWrittenAgainAfterBeingTakenAwayIsAskedAgainUnderANewAsking(t *testing.T) {
|
func anAssignerAskWrittenAgainAfterBeingTakenAwayIsAskedAgainUnderANewAsking(t *testing.T) {
|
||||||
digest := "sha256:" + strings.Repeat("e", 64)
|
digest := "sha256:" + strings.Repeat("e", 64)
|
||||||
fp := "src2:" + strings.Repeat("a", 64)
|
fp := "src2:" + strings.Repeat("a", 64)
|
||||||
recordKey, _ := asks.RecordAskKey("novox", "nats", digest)
|
recordKey, _ := asks.RecordAskKey("novox", "nats", digest)
|
||||||
restoreKey, _ := asks.RestoreAskKey("novox", "nats", fp)
|
restoreKey, _ := asks.RestoreAskKey("novox", "nats", fp)
|
||||||
holdKey, _ := asks.HoldAskKey("nats", fp)
|
holdKey, _ := asks.HoldAskKey("nats", fp)
|
||||||
for kind, key := range map[string]string{"record": recordKey, "restore": restoreKey, "hold": holdKey} {
|
backupKey, _ := asks.BackupAskKey("novox", "nats", fp)
|
||||||
t.Run(kind, func(t *testing.T) {
|
for kind, key := range map[string]string{"record": recordKey, "restore": restoreKey, "hold": holdKey, "backup": backupKey} {
|
||||||
write := func(assigner memGate, asking int) oc.Ask {
|
write := func(t *testing.T, assigner memGate, asking int) oc.Ask {
|
||||||
ask := oc.Ask{ID: meshdeliveryclient.AskingID(key, asking), Headline: "nats waits", Explanation: "e", Who: occ.Operator,
|
ask := oc.Ask{ID: meshdeliveryclient.AskingID(key, asking), Headline: "nats waits", Explanation: "e", Who: occ.Operator,
|
||||||
Expires: occ.FormatTime(now.Add(occ.ApproveLasts)), OnExpiry: "stays",
|
Expires: occ.FormatTime(now.Add(occ.ApproveLasts)), OnExpiry: "stays",
|
||||||
Options: []oc.Option{{ID: "yes", Label: "Yes", Does: "d", Level: occ.Acknowledge}, {ID: "no", Label: "No", Does: "d", Level: occ.Acknowledge}}}
|
Options: []oc.Option{{ID: "yes", Label: "Yes", Does: "d", Level: occ.Acknowledge}, {ID: "no", Label: "No", Does: "d", Level: occ.Acknowledge}}}
|
||||||
value, err := asks.EncodeAsk(key, asks.Ask{Ask: rawAsk(t, ask), Asking: asking})
|
value, err := asks.EncodeAsk(key, asks.Ask{Ask: rawAsk(t, ask), Asking: asking})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
|
||||||
assigner[key] = value
|
|
||||||
return ask
|
|
||||||
}
|
}
|
||||||
|
assigner[key] = value
|
||||||
|
return ask
|
||||||
|
}
|
||||||
|
t.Run(kind, func(t *testing.T) {
|
||||||
b := newBed(t)
|
b := newBed(t)
|
||||||
assigner, _ := withView(b)
|
assigner, _ := withView(b)
|
||||||
first := write(assigner, 1000)
|
first := write(t, assigner, 1000)
|
||||||
b.t.OnAssignerAsk(key)
|
b.t.OnAssignerAsk(key)
|
||||||
delete(assigner, key)
|
delete(assigner, key)
|
||||||
b.t.OnAssignerAskGone(key) // the assigner took it away as stale: the router ask is taken back
|
b.t.OnAssignerAskGone(key) // the assigner took it away as stale: the router ask is taken back
|
||||||
second := write(assigner, 1060)
|
second := write(t, assigner, 1060)
|
||||||
b.t.OnAssignerAsk(key)
|
b.t.OnAssignerAsk(key)
|
||||||
if got := b.askedIDs(); len(got) != 2 || got[0] != first.ID || got[1] != second.ID || first.ID == second.ID {
|
if got := b.askedIDs(); len(got) != 2 || got[0] != first.ID || got[1] != second.ID || first.ID == second.ID {
|
||||||
t.Fatalf("asked %v, want %s then %s; log:\n%s", got, first.ID, second.ID, b.logs.String())
|
t.Fatalf("asked %v, want %s then %s; log:\n%s", got, first.ID, second.ID, b.logs.String())
|
||||||
}
|
}
|
||||||
// Written again under the id the router kept, after it was taken back: not asked, and said.
|
if len(b.emitted) != 0 {
|
||||||
|
t.Fatalf("said %v while every asking was asked", b.emitted)
|
||||||
|
}
|
||||||
|
// Written again under the id the router kept, after it was taken back: not asked, and the assigner is told.
|
||||||
delete(assigner, key)
|
delete(assigner, key)
|
||||||
|
b.t.OnAssignerAsk(key) // a watch's replay, as at the teller's start, before the key went
|
||||||
b.t.OnAssignerAskGone(key)
|
b.t.OnAssignerAskGone(key)
|
||||||
write(assigner, 1060)
|
write(t, assigner, 1060)
|
||||||
b.t.OnAssignerAsk(key)
|
b.t.OnAssignerAsk(key)
|
||||||
if got := b.askedIDs(); len(got) != 2 || !strings.Contains(b.logs.String(), "wrote the key again under the same id") {
|
want := []string{tellerevents.AskEnded + " " + key + " " + second.ID + " " + occ.OutcomeCancelled}
|
||||||
t.Errorf("asked %v after the same id was written again; log:\n%s", got, b.logs.String())
|
if got := b.askedIDs(); len(got) != 2 || !slices.Equal(b.emitted, want) {
|
||||||
|
t.Errorf("asked %v and said %v after the same id was written again, want %v; log:\n%s", got, b.emitted, want, b.logs.String())
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
for _, end := range []struct {
|
||||||
|
name string
|
||||||
|
end func(b *bed, a oc.Ask)
|
||||||
|
said string // the outcome said to the assigner, or "" for none
|
||||||
|
}{
|
||||||
|
{"lapsed at the router", func(b *bed, a oc.Ask) {
|
||||||
|
b.deliver(t, oc.Warrant{Ask: a.ID, Asker: tellerModule, Outcome: occ.OutcomeExpired, At: occ.FormatTime(now), AskDigest: occ.Digest(a)})
|
||||||
|
}, occ.OutcomeExpired},
|
||||||
|
{"cancelled at the router", func(b *bed, a oc.Ask) {
|
||||||
|
b.deliver(t, oc.Warrant{Ask: a.ID, Asker: tellerModule, Outcome: occ.OutcomeCancelled, At: occ.FormatTime(now), AskDigest: occ.Digest(a)})
|
||||||
|
}, occ.OutcomeCancelled},
|
||||||
|
{"replaced at the router", func(b *bed, a oc.Ask) {
|
||||||
|
b.deliver(t, oc.Warrant{Ask: a.ID, Asker: tellerModule, Outcome: occ.OutcomeReplaced, At: occ.FormatTime(now), AskDigest: occ.Digest(a)})
|
||||||
|
}, ""},
|
||||||
|
{"refused for its words", func(b *bed, a oc.Ask) { b.refuse(t, a.ID, "the ask has no headline") }, ""},
|
||||||
|
{"answered", func(b *bed, a oc.Ask) { b.deliver(t, ackFor(a, "yes")) }, ""},
|
||||||
|
{"ended by hand", func(b *bed, a oc.Ask) { b.t.EndAsk(a.ID, "the operator asked", "lead@shanks") }, ""},
|
||||||
|
} {
|
||||||
|
t.Run(kind+"/"+end.name, func(t *testing.T) {
|
||||||
|
b := newBed(t)
|
||||||
|
assigner, _ := withView(b)
|
||||||
|
a := write(t, assigner, 1000)
|
||||||
|
b.t.OnAssignerAsk(key)
|
||||||
|
end.end(b, a)
|
||||||
|
b.t.OnAssignerAsk(key) // the key still stands under that asking, as at the teller's start
|
||||||
|
var want []string
|
||||||
|
if end.said != "" {
|
||||||
|
want = []string{tellerevents.AskEnded + " " + key + " " + a.ID + " " + end.said}
|
||||||
|
}
|
||||||
|
if !slices.Equal(b.emitted, want) {
|
||||||
|
t.Errorf("said %v, want %v; log:\n%s", b.emitted, want, b.logs.String())
|
||||||
|
}
|
||||||
|
if got := b.askedIDs(); len(got) != 1 {
|
||||||
|
t.Errorf("asked %v, want the one asking only: the assigner writes a new one", got)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -63,6 +63,7 @@ func (t *Teller) EndAsk(ask, why, by string) ends.Result {
|
|||||||
r.Says = "the ask " + id + " could not be cancelled at the router (see the teller's journal): it stays open"
|
r.Says = "the ask " + id + " could not be cancelled at the router (see the teller's journal): it stays open"
|
||||||
return r
|
return r
|
||||||
}
|
}
|
||||||
|
t.markHand(key, id)
|
||||||
r.Ended = true
|
r.Ended = true
|
||||||
r.Says = "the ask " + id + " is cancelled at the router, which frees its place; the next ask waiting for room is asked when the router says it ended"
|
r.Says = "the ask " + id + " is cancelled at the router, which frees its place; the next ask waiting for room is asked when the router says it ended"
|
||||||
t.logf("%s: the ask %s is ended by hand, by %s: %s", key, id, by, why)
|
t.logf("%s: the ask %s is ended by hand, by %s: %s", key, id, by, why)
|
||||||
@@ -90,3 +91,16 @@ func endAsked(ctx context.Context, t *Teller, kv jetstream.KeyValue, key string,
|
|||||||
t.logf("%s.%s: the answer could not be written: %v", ends.State, key, err)
|
t.logf("%s.%s: the answer could not be written: %v", ends.State, key, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// markHand keeps, beside the latest asking of key, that a person ended the asking id by hand, so the assigner is not told it
|
||||||
|
// ended at the channel (issue 1966). A record that names another asking is left.
|
||||||
|
func (t *Teller) markHand(key, id string) {
|
||||||
|
g, rec, err := t.seenAt(gateRecordKey(key))
|
||||||
|
if err != nil || rec == nil || g.ID != id || g.Hand {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
g.Hand = true
|
||||||
|
if _, err := t.Store.Update(gateRecordKey(key), g, rec.Revision); err != nil {
|
||||||
|
t.logf("%s: that the ask %s was ended by hand could not be kept: %v", key, id, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -79,7 +79,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
|||||||
client := &occ.Client{Module: tellerModule, State: state, Publish: publish,
|
client := &occ.Client{Module: tellerModule, State: state, Publish: publish,
|
||||||
Record: recordVia(ctx, conn, tellerModule), Subscribe: decidedVia(hub.Handle, logger.Printf)}
|
Record: recordVia(ctx, conn, tellerModule), Subscribe: decidedVia(hub.Handle, logger.Printf)}
|
||||||
t := &Teller{Asker: client, Store: state, Answers: kvAnswers{ctx: ctx, kv: kvs[0]}, Gate: kvGate{ctx: ctx, kv: gate},
|
t := &Teller{Asker: client, Store: state, Answers: kvAnswers{ctx: ctx, kv: kvs[0]}, Gate: kvGate{ctx: ctx, kv: gate},
|
||||||
Say: sayVia(ctx, js, tellerModule), Logf: logger.Printf, Publish: publish}
|
Say: sayVia(ctx, js, tellerModule), Logf: logger.Printf, Publish: publish,
|
||||||
|
// The teller's own events (its manifest's emits), on EVENTS under its module's subject.
|
||||||
|
Emit: func(event string, body []byte, id string) error {
|
||||||
|
_, err := js.Publish(ctx, cred.EventSubject(event), body, jetstream.WithMsgID(id))
|
||||||
|
return err
|
||||||
|
}}
|
||||||
// Hear the warrants first, so none that comes while the asks are read is missed.
|
// Hear the warrants first, so none that comes while the asks are read is missed.
|
||||||
if err := client.Listen(t.Answer, t.Said); err != nil {
|
if err := client.Listen(t.Answer, t.Said); err != nil {
|
||||||
return fmt.Errorf("hearing the router's warrants: %w", err)
|
return fmt.Errorf("hearing the router's warrants: %w", err)
|
||||||
|
|||||||
@@ -102,6 +102,9 @@ type Teller struct {
|
|||||||
Publish func(subject string, body any, id string) error
|
Publish func(subject string, body any, id string) error
|
||||||
// Now is the clock; nil is time.Now.
|
// Now is the clock; nil is time.Now.
|
||||||
Now func() time.Time
|
Now func() time.Time
|
||||||
|
// Emit says one of the teller's own events (its manifest's emits, tellerevents) under a bus message id of its own; nil
|
||||||
|
// says none, and the journal says so.
|
||||||
|
Emit func(event string, body []byte, id string) error
|
||||||
|
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
}
|
}
|
||||||
@@ -119,6 +122,9 @@ type seen struct {
|
|||||||
Key string `json:"key,omitempty"`
|
Key string `json:"key,omitempty"`
|
||||||
ID string `json:"id,omitempty"`
|
ID string `json:"id,omitempty"`
|
||||||
Rev uint64 `json:"rev"`
|
Rev uint64 `json:"rev"`
|
||||||
|
// Hand marks the asking ID as ended by hand (teller_end_ask): a person's end, so the assigner is never told it ended at
|
||||||
|
// the channel and its key is not written again for it (issue 1966).
|
||||||
|
Hand bool `json:"hand,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
func gateRecordKey(key string) string { return "gate." + key }
|
func gateRecordKey(key string) string { return "gate." + key }
|
||||||
@@ -454,6 +460,11 @@ func (t *Teller) said(h occ.Handled, w oc.Warrant, err error) {
|
|||||||
return // a second word of the same refusal is no end of an ask
|
return // a second word of the same refusal is no end of an ask
|
||||||
}
|
}
|
||||||
t.saidOnce(h, w, err)
|
t.saidOnce(h, w, err)
|
||||||
|
if h == occ.Recorded {
|
||||||
|
if s, rec, _ := t.seenOf(w.Ask); rec != nil && isAssignerKey(s.Key) {
|
||||||
|
t.sayAssignerAskEnded(s.Key, w.Ask, w.Outcome)
|
||||||
|
}
|
||||||
|
}
|
||||||
if h == occ.Acted || h == occ.Recorded || h == occ.Done {
|
if h == occ.Acted || h == occ.Recorded || h == occ.Done {
|
||||||
// One of the teller's asks ended at the router, so it has room for one more: the first ask waiting is asked again.
|
// One of the teller's asks ended at the router, so it has room for one more: the first ask waiting is asked again.
|
||||||
t.endWait(w.Ask)
|
t.endWait(w.Ask)
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
|
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"
|
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
||||||
|
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
@@ -132,6 +133,8 @@ type bed struct {
|
|||||||
busDropped []string
|
busDropped []string
|
||||||
hear func(stdio.Envelope) error
|
hear func(stdio.Envelope) error
|
||||||
logs strings.Builder
|
logs strings.Builder
|
||||||
|
// emitted are the teller's own events said, as `<event> <key> <ask> <outcome>`.
|
||||||
|
emitted []string
|
||||||
}
|
}
|
||||||
|
|
||||||
func newBed(t *testing.T) *bed {
|
func newBed(t *testing.T) *bed {
|
||||||
@@ -178,7 +181,20 @@ func newBed(t *testing.T) *bed {
|
|||||||
b.conds = append(b.conds, c)
|
b.conds = append(b.conds, c)
|
||||||
return nil
|
return nil
|
||||||
},
|
},
|
||||||
Logf: func(f string, a ...any) { fmt.Fprintf(&b.logs, f+"\n", a...) }}
|
Logf: func(f string, a ...any) { fmt.Fprintf(&b.logs, f+"\n", a...) },
|
||||||
|
Emit: func(event string, body []byte, id string) error {
|
||||||
|
e, err := tellerevents.DecodeEnded(body)
|
||||||
|
if err != nil || id != event+"."+e.Ask {
|
||||||
|
t.Errorf("the event %s said under %s holds %s: %v", event, id, body, err)
|
||||||
|
}
|
||||||
|
if b.msgIDs[id] {
|
||||||
|
b.busDropped = append(b.busDropped, id) // the bus drops a second message under an id it holds
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
b.msgIDs[id] = true
|
||||||
|
b.emitted = append(b.emitted, strings.Join([]string{event, e.Key, e.Ask, e.Outcome}, " "))
|
||||||
|
return nil
|
||||||
|
}}
|
||||||
if err := b.client.Listen(b.t.Answer, b.t.Said); err != nil {
|
if err := b.client.Listen(b.t.Answer, b.t.Said); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ require (
|
|||||||
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient v0.0.0-00010101000000-000000000000
|
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient v0.0.0-00010101000000-000000000000
|
||||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000
|
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000
|
||||||
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks v0.0.0-00010101000000-000000000000
|
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks v0.0.0-00010101000000-000000000000
|
||||||
|
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents v0.0.0-00010101000000-000000000000
|
||||||
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate v0.0.0-00010101000000-000000000000
|
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate v0.0.0-00010101000000-000000000000
|
||||||
git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 // indirect
|
git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 // indirect
|
||||||
git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000
|
git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000
|
||||||
@@ -35,6 +36,8 @@ require (
|
|||||||
|
|
||||||
replace git.novox.be/novox/mesh-delivery/sdk => ../../sdk
|
replace git.novox.be/novox/mesh-delivery/sdk => ../../sdk
|
||||||
|
|
||||||
|
replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents => ./tellerevents
|
||||||
|
|
||||||
require (
|
require (
|
||||||
git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000 // indirect
|
git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000 // indirect
|
||||||
git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000 // indirect
|
git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000 // indirect
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package manifest
|
|||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents"
|
||||||
"io/fs"
|
"io/fs"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
@@ -42,6 +43,7 @@ type manifest struct {
|
|||||||
Reads []string `json:"reads"`
|
Reads []string `json:"reads"`
|
||||||
Invokes []string `json:"invokes"`
|
Invokes []string `json:"invokes"`
|
||||||
Consumes []string `json:"consumes"`
|
Consumes []string `json:"consumes"`
|
||||||
|
Emits []string `json:"emits"`
|
||||||
OwnSecrets map[string]string `json:"own-secrets"`
|
OwnSecrets map[string]string `json:"own-secrets"`
|
||||||
SecretsOwner string `json:"secrets-owner"`
|
SecretsOwner string `json:"secrets-owner"`
|
||||||
Resources []map[string]any `json:"resources"`
|
Resources []map[string]any `json:"resources"`
|
||||||
@@ -116,8 +118,12 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1
|
|||||||
if want := []string{"teller_end_ask", "teller_first_views", "teller_first_view_again"}; !reflect.DeepEqual(m.Tools, want) {
|
if want := []string{"teller_end_ask", "teller_first_views", "teller_first_view_again"}; !reflect.DeepEqual(m.Tools, want) {
|
||||||
t.Errorf("tools = %v, want %v", m.Tools, want)
|
t.Errorf("tools = %v, want %v", m.Tools, want)
|
||||||
}
|
}
|
||||||
// No emits, streams, bus or seat (ADR 0437 §4).
|
// Its one event: an assigner ask ended at the channel without an answer, heard by the assigner (issue 1966, tellerevents).
|
||||||
for _, k := range []string{"streams", "logs", "emits", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus"} {
|
if want := []string{tellerevents.AskEnded}; !reflect.DeepEqual(m.Emits, want) {
|
||||||
|
t.Errorf("emits = %v, want %v", m.Emits, want)
|
||||||
|
}
|
||||||
|
// No streams, bus or seat (ADR 0437 §4).
|
||||||
|
for _, k := range []string{"streams", "logs", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus"} {
|
||||||
if _, ok := top[k]; ok {
|
if _, ok := top[k]; ok {
|
||||||
t.Errorf("the manifest declares %q: the teller has none", k)
|
t.Errorf("the manifest declares %q: the teller has none", k)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -14,6 +14,9 @@
|
|||||||
"consumes": [
|
"consumes": [
|
||||||
"mesh-delivery-assigner.views.asks"
|
"mesh-delivery-assigner.views.asks"
|
||||||
],
|
],
|
||||||
|
"emits": [
|
||||||
|
"ask-ended"
|
||||||
|
],
|
||||||
"state": [
|
"state": [
|
||||||
{ "name": "answers", "history": 5 },
|
{ "name": "answers", "history": 5 },
|
||||||
{ "name": "asked", "history": 1 },
|
{ "name": "asked", "history": 1 },
|
||||||
|
|||||||
@@ -0,0 +1,5 @@
|
|||||||
|
module git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents
|
||||||
|
|
||||||
|
go 1.26.0
|
||||||
|
|
||||||
|
require git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000
|
||||||
@@ -0,0 +1,59 @@
|
|||||||
|
// Package tellerevents holds the layouts of the teller's own events, which other modules consume (novox/hq ADR 0634 §1: a
|
||||||
|
// layout one module owns is that module's own package). The teller says them as its module's events,
|
||||||
|
// `mesh.mod.mesh-delivery-teller.event.<name>`, named in its manifest's `emits`.
|
||||||
|
package tellerevents
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
|
||||||
|
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||||
|
)
|
||||||
|
|
||||||
|
// AskEnded is the event the teller says when one of the assigner's asks ended at the operator channel without an answer
|
||||||
|
// (lapsed, or cancelled there) while its key still stands under that asking (issue 1966: the nats yes on novox and three
|
||||||
|
// backup asks were asked once, ended at the channel, and nothing asked them again). The assigner hears it and writes the
|
||||||
|
// key again under a new asking where the ask's kind allows it.
|
||||||
|
const AskEnded = "ask-ended"
|
||||||
|
|
||||||
|
// Ended is the body of AskEnded: the key of the ask in the assigner's asks, the router id of the asking that ended, and the
|
||||||
|
// router's outcome (expired or cancelled).
|
||||||
|
type Ended struct {
|
||||||
|
V int `json:"v"`
|
||||||
|
Key string `json:"key"`
|
||||||
|
Ask string `json:"ask"`
|
||||||
|
Outcome string `json:"outcome"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func (e Ended) check() error {
|
||||||
|
for _, f := range []struct{ name, value string }{{"key", e.Key}, {"ask", e.Ask}, {"outcome", e.Outcome}} {
|
||||||
|
if f.value == "" {
|
||||||
|
return fmt.Errorf("tellerevents: an ended ask has no %s", f.name)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// EncodeEnded is e as the body of AskEnded, at keys.Version; it refuses an ended ask lacking its key, its ask or its outcome.
|
||||||
|
func EncodeEnded(e Ended) ([]byte, error) {
|
||||||
|
if err := e.check(); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
e.V = keys.Version
|
||||||
|
return json.Marshal(e)
|
||||||
|
}
|
||||||
|
|
||||||
|
// DecodeEnded reads the body of AskEnded; it refuses an unknown version, an unknown field and what EncodeEnded refuses.
|
||||||
|
func DecodeEnded(data []byte) (Ended, error) {
|
||||||
|
if err := keys.CheckVersion(data); err != nil {
|
||||||
|
return Ended{}, err
|
||||||
|
}
|
||||||
|
var e Ended
|
||||||
|
dec := json.NewDecoder(bytes.NewReader(data))
|
||||||
|
dec.DisallowUnknownFields()
|
||||||
|
if err := dec.Decode(&e); err != nil {
|
||||||
|
return Ended{}, fmt.Errorf("tellerevents: an ended ask does not fit its shape: %w", err)
|
||||||
|
}
|
||||||
|
return e, e.check()
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user