diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/kb.go b/modules/mesh-knowledge/cmd/mesh-knowledge/kb.go index 13deaa96..80d13542 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/kb.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/kb.go @@ -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() diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore.go b/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore.go index 97985c51..909920c7 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/ledgerstore.go @@ -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 + } +} diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/main.go b/modules/mesh-knowledge/cmd/mesh-knowledge/main.go index 85122f86..cc1f5c9f 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/main.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/main.go @@ -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() diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/sent.go b/modules/mesh-knowledge/cmd/mesh-knowledge/sent.go new file mode 100644 index 00000000..7eb84a97 --- /dev/null +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/sent.go @@ -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 +} diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/sent_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/sent_test.go new file mode 100644 index 00000000..e49cf069 --- /dev/null +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/sent_test.go @@ -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:] } diff --git a/modules/mesh-knowledge/cmd/mesh-knowledge/writing_test.go b/modules/mesh-knowledge/cmd/mesh-knowledge/writing_test.go index 914b9db6..29b2c904 100644 --- a/modules/mesh-knowledge/cmd/mesh-knowledge/writing_test.go +++ b/modules/mesh-knowledge/cmd/mesh-knowledge/writing_test.go @@ -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 }