CAT-K7: every operation's event is sent once, after a restart too

An event of a change carries the record, the version it follows and the event as its message id; the last version sent per record is a key of the state sent; at start the events of the operations after it are sent again under the same ids, so a stop between an operation and its event loses none and a second start sends none. Issue 1966. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 12:09:26 +01:00
parent c5dc10fa8c
commit 7bf0e991d8
6 changed files with 255 additions and 19 deletions
@@ -88,14 +88,16 @@ type Clearer func(key, resolved string) error

 // KB is the knowledge base.
 type KB struct {
-	Store    Store
-	Records  Records
-	Reserve  Reserver
-	Judge    Judgement
-	Asks     AskBus
-	Notify   Notifier
-	Clear    Clearer
-	Emit     func(event string, body any) error
+	Store   Store
+	Records Records
+	Reserve Reserver
+	Judge   Judgement
+	Asks    AskBus
+	Notify  Notifier
+	Clear   Clearer
+	Emit    func(event string, body any) error
+	// EmitID says an event under a message id of its own, which the bus drops a repeat of.
+	EmitID   func(event string, body any, id string) error
 	Settings func() (Settings, error)
 	Now      func() time.Time
 	Logf     func(format string, a ...any)
@@ -160,11 +162,28 @@ func (k *KB) emit(event string, r *Record, by, what string, version int) {
 	}
 	body := map[string]any{"kind": r.Kind, "number": r.Number, "record": r.Key().String(), "by": by, "at": k.now(),
 		"what": what, "version": version, "status": r.Status}
-	if err := k.Emit(event, body); err != nil {
+	if err := k.send(event, r.Key(), version, body); err != nil {
 		k.logf("[mesh-knowledge] the event %s for %s could not be said: %v", event, r.Key(), err)
 	}
 }

+// send says an event once: its message id is the record, the version (the operation it follows) and the event, so the
+// bus drops a second send within its duplicate window (to-be 55 §1); the last version sent per record is kept in the state
+// `sent`, and the events of later operations are sent at start (ResendEvents), so a module stopped between an operation
+// and its event loses none. An event of no version (a section written) has no id of its own.
+func (k *KB) send(event string, key Key, version int, body map[string]any) error {
+	var err error
+	if k.EmitID != nil && version > 0 {
+		err = k.EmitID(event, body, fmt.Sprintf("%s.%04d.v%d.%s", key.Kind, key.Number, version, event))
+	} else {
+		err = k.Emit(event, body)
+	}
+	if err == nil && version > 0 {
+		k.markSent(key, version)
+	}
+	return err
+}
+
 // reference is the reading order and the glossary, read again after ten minutes.
 func (k *KB) reference() (*Reference, error) {
 	k.refMu.Lock()
@@ -312,6 +312,8 @@ func operationFor(old *Entry, v Version, current bool) string {
 	}
 	prev, next := old.Working(), v.Record
 	switch {
+	case prev.Status == StatusReserved && next.Status == StatusProposed:
+		return OpPropose
 	case next.Status == StatusWithdrawn && prev.Status != StatusWithdrawn:
 		return OpWithdraw
 	case next.Accepted() && !prev.Accepted(), next.Status == StatusSuperseded && prev.Status != StatusSuperseded:
@@ -567,3 +569,29 @@ func (s *ledgerStore) Untold(_ context.Context, key string) error {
 }

 func isChanged(err error) bool { return errors.Is(err, stdio.ErrChanged) }
+
+// Operations are the operations of a record that made a version (OperationReader).
+func (s *ledgerStore) Operations(_ context.Context, k Key) ([]OperationInfo, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	var out []OperationInfo
+	for from := uint64(0); ; {
+		var page buscalls.HistoryPage
+		if err := s.pipe.Call(buscalls.LedgerHistory, buscalls.HistoryRequest{Ledger: LedgerName, Instance: InstanceOf(k), From: from}, &page); err != nil {
+			return nil, err
+		}
+		for _, op := range page.Operations {
+			var body struct {
+				Version *versionHead `json:"version"`
+			}
+			if op.Operation == OpWriteSection || json.Unmarshal(op.Body, &body) != nil || body.Version == nil {
+				continue
+			}
+			out = append(out, OperationInfo{Sequence: op.Sequence, Operation: op.Operation, Version: body.Version.N, By: body.Version.By, At: body.Version.At, Why: body.Version.Why})
+		}
+		if page.Next == 0 {
+			return out, nil
+		}
+		from = page.Next
+	}
+}
@@ -81,15 +81,18 @@ func main() {
 func newKB(store Store) *KB {
 	file := os.Getenv("MESH_KNOWLEDGE_SETTINGS")
 	return &KB{
-		Store:    store,
-		Records:  busRecords{},
-		Seats:    stdio.Ask,
-		Reserve:  busReserver,
-		Judge:    busJudgement(os.Getenv("MESH_NODE")),
-		Asks:     sdkAskBus{},
-		Notify:   busNotifier,
-		Clear:    busClearer,
-		Emit:     func(event string, body any) error { return stdio.Emit(event, body) },
+		Store:   store,
+		Records: busRecords{},
+		Seats:   stdio.Ask,
+		Reserve: busReserver,
+		Judge:   busJudgement(os.Getenv("MESH_NODE")),
+		Asks:    sdkAskBus{},
+		Notify:  busNotifier,
+		Clear:   busClearer,
+		Emit:    func(event string, body any) error { return stdio.Emit(event, body) },
+		EmitID: func(event string, body any, id string) error {
+			return stdio.Emit(event, body, stdio.EmitOptions{MessageID: id})
+		},
 		Settings: func() (Settings, error) { return readSettings(file) },
 		Logf:     logf,
 		States:   sdkStates{},
@@ -139,6 +142,12 @@ func background(k *KB) {
 	tick := time.NewTicker(time.Minute)
 	defer tick.Stop()
 	time.Sleep(5 * time.Second)
+	// The events of the operations a stop lost are sent first (to-be 55 §1).
+	if n, err := k.ResendEvents(context.Background()); err != nil {
+		logf("[mesh-knowledge] sending the events a stop lost: %v", err)
+	} else if n > 0 {
+		logf("[mesh-knowledge] sent %d event(s) a stop lost", n)
+	}
 	review := func() {
 		for {
 			said, err := k.ReviewStep()
@@ -0,0 +1,120 @@
+package main
+
+// Every operation's event is sent once, after a restart too (novox/hq to-be 55 §1): the event of a change carries the
+// version it follows as its message id, so the bus drops a second send; the last version sent per record is a key of the
+// state `sent`; and at start the events of every later operation are sent, so a module stopped between an operation and its
+// event loses none, and a consumer of `reviewed` never waits for one that was never sent.
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+)
+
+const sentState = "sent"
+
+func sentKey(k Key) string { return fmt.Sprintf("%s.%04d", k.Kind, k.Number) }
+
+// markSent keeps that the events up to version were sent. It only moves forward.
+func (k *KB) markSent(key Key, version int) {
+	if k.States == nil {
+		return
+	}
+	kv := k.States.State(sentState)
+	for try := 0; try < 3; try++ {
+		e, err := kv.Get(sentKey(key))
+		if err != nil {
+			k.logf("[mesh-knowledge] what was sent of %s cannot be read: %v", key, err)
+			return
+		}
+		if e == nil {
+			if _, err = kv.Create(sentKey(key), version); err == nil {
+				return
+			}
+			continue
+		}
+		var have int
+		_ = json.Unmarshal(e.Value, &have)
+		if have >= version {
+			return
+		}
+		if _, err = kv.Update(sentKey(key), version, e.Revision); err == nil {
+			return
+		}
+	}
+}
+
+// sentUpTo is the last version of a record whose events were sent.
+func (k *KB) sentUpTo(key Key) int {
+	if k.States == nil {
+		return 0
+	}
+	e, err := k.States.State(sentState).Get(sentKey(key))
+	if err != nil || e == nil {
+		return 0
+	}
+	var have int
+	_ = json.Unmarshal(e.Value, &have)
+	return have
+}
+
+// OperationInfo is one operation of a record's history, as far as sending its event needs.
+type OperationInfo struct {
+	Sequence  uint64
+	Operation string
+	Version   int
+	By        string
+	At        time.Time
+	Why       string
+}
+
+// OperationReader is a store that keeps every change as an operation: the ledger.
+type OperationReader interface {
+	// Operations are the operations of a record that made a version, oldest first.
+	Operations(ctx context.Context, k Key) ([]OperationInfo, error)
+}
+
+// eventOf is the event an operation says; none for an operation whose change is not an event of its own.
+var eventOf = map[string]string{OpReserve: EventReserved, OpPropose: EventProposed, OpImport: EventImported, OpReview: EventReviewed,
+	OpAccept: EventAccepted, OpSubmit: EventSubmitted, OpTakeOver: EventTakenOver}
+
+// ResendEvents sends, at start, the event of every operation after the last one sent of each record, and answers how
+// many it sent. The same message id as the first send, so a send that did arrive is dropped by the bus.
+func (k *KB) ResendEvents(ctx context.Context) (int, error) {
+	r, ok := k.Store.(OperationReader)
+	if !ok || k.EmitID == nil {
+		return 0, nil
+	}
+	all, err := k.Store.All(ctx)
+	if err != nil {
+		return 0, err
+	}
+	sent := 0
+	for _, e := range all {
+		key := e.Current.Key()
+		from := k.sentUpTo(key)
+		if from >= e.Versions {
+			continue
+		}
+		ops, err := r.Operations(ctx, key)
+		if err != nil {
+			return sent, err
+		}
+		for _, op := range ops {
+			if op.Version <= from {
+				continue
+			}
+			if event := eventOf[op.Operation]; event != "" {
+				body := map[string]any{"kind": key.Kind, "number": key.Number, "record": key.String(), "by": op.By, "at": op.At,
+					"what": op.Why, "version": op.Version, "status": e.Current.Status}
+				if err := k.send(event, key, op.Version, body); err != nil {
+					return sent, fmt.Errorf("the event %s for %s could not be sent: %v", event, key, err)
+				}
+				sent++
+			}
+		}
+		k.markSent(key, e.Versions)
+	}
+	return sent, nil
+}
@@ -0,0 +1,57 @@
+package main
+
+import (
+	"errors"
+	"strings"
+	"testing"
+)
+
+// Every operation's event is sent once, after a restart too (to-be 55 §1, AC51): the message id is the record, the version
+// and the event; the last version sent is kept; a module stopped between an operation and its event sends it at start,
+// under the same id, and sends it once.
+func TestEveryOperationsEventIsSentOnceAfterARestartToo(t *testing.T) {
+	w, f := ledgerWorld(t)
+	type send struct{ event, id string }
+	var sends []send
+	down := false
+	w.k.EmitID = func(event string, body any, id string) error {
+		if down {
+			return errors.New("the bus is down")
+		}
+		sends = append(sends, send{event, id})
+		return nil
+	}
+	r, err := w.k.ReserveNumber("author@g14", KindDecision, "a test record")
+	if err != nil {
+		t.Fatal(err)
+	}
+	key := Key{KindDecision, r.Number}
+	if len(sends) != 1 || sends[0].event != EventReserved || sends[0].id != key.Kind+"."+strings.Repeat("0", 4-len(itoa(key.Number)))+itoa(key.Number)+".v1.reserved" {
+		t.Fatalf("%+v", sends)
+	}
+	if k := w.k.sentUpTo(key); k != 1 {
+		t.Fatalf("sent up to %d", k)
+	}
+	// Stopped between the operation and its event: the proposal is in the history, its event was not sent.
+	down = true
+	if _, err := w.k.Propose("author@g14", valid(r.Number)); err != nil {
+		t.Fatal(err)
+	}
+	if w.k.sentUpTo(key) != 1 {
+		t.Fatal("an event never sent was marked sent")
+	}
+	// A restart: a new knowledge base over the same ledger and states sends what the stop lost, once.
+	down = false
+	sends = nil
+	again := &KB{Store: w.k.Store, States: w.k.States, EmitID: w.k.EmitID, Now: w.k.Now, Logf: w.k.Logf}
+	n, err := again.ResendEvents(t.Context())
+	if err != nil || n != 1 || len(sends) != 1 || sends[0].event != EventProposed || !strings.HasSuffix(sends[0].id, ".v2.proposed") {
+		t.Fatalf("%v: %d, %+v", err, n, sends)
+	}
+	if n, _ := again.ResendEvents(t.Context()); n != 0 || len(sends) != 1 {
+		t.Fatalf("a second start sent %d again", n)
+	}
+	_ = f
+}
+
+func itoa(n int) string { return Key{KindDecision, n}.String()[4:] }
@@ -9,8 +9,11 @@ import (
 func ledgerWorld(t *testing.T) (*world, *fakeLedger) {
 	t.Helper()
 	f := newFakeLedger(t)
-	w := imported(t, newLedgerStore(f, newFakeStates()))
+	w := newWorld(t, newLedgerStore(f, newFakeStates()), fixture())
 	w.k.States = newFakeStates()
+	if _, err := w.k.Import("import@test", ImportAsk{}, fixture()); err != nil {
+		t.Fatal(err)
+	}
 	return w, f
 }