diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/bridge_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/bridge_test.go index fb246c590..13770632f 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/bridge_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/bridge_test.go @@ -115,6 +115,47 @@ func (m *fakeMerges) last(group string) Merge { return m.values[group] } +// fakeOld is the gate's old states merges and owed, which only the switch reads. +type fakeOld struct { + mu sync.Mutex + states map[string]map[string]oldEntry +} + +type oldEntry struct { + raw []byte + revision uint64 +} + +func (o *fakeOld) put(state, key string, raw []byte, revision uint64) { + o.mu.Lock() + defer o.mu.Unlock() + if o.states[state] == nil { + o.states[state] = map[string]oldEntry{} + } + o.states[state][key] = oldEntry{raw: raw, revision: revision} +} + +func (o *fakeOld) Keys(state string) ([]string, error) { + o.mu.Lock() + defer o.mu.Unlock() + var out []string + for k := range o.states[state] { + out = append(out, k) + } + sort.Strings(out) + return out, nil +} + +func (o *fakeOld) Get(state, key string) ([]byte, uint64, error) { + o.mu.Lock() + defer o.mu.Unlock() + e, ok := o.states[state][key] + if !ok { + return nil, 0, nil + } + return e.raw, e.revision, nil +} + // fakeSource is the gate's bucket source with its one key. type fakeSource struct { mu sync.Mutex @@ -174,6 +215,12 @@ func newFakeOwed(f *fakeForge) *fakeOwed { } func (o *fakeOwed) Say(v seat.Owed) error { + if _, _, _, switched := mdc.ParseSwitchedActID(v.ID); switched { + o.mu.Lock() + defer o.mu.Unlock() + o.said = append(o.said, v) // the bridge counts it against what the old bridge performed + return nil + } key, _, found := strings.Cut(v.ID, "@") if !found { return fmt.Errorf("the act's id %q holds no @", v.ID) diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/fakes_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/fakes_test.go index d966a7b8d..4271e3165 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/fakes_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/fakes_test.go @@ -287,6 +287,7 @@ type world struct { forge *fakeForge op *fakeOperator merges *fakeMerges + old *fakeOld source *fakeSource owed *fakeOwed // asked is the key of the ask a test answered. @@ -320,7 +321,8 @@ func newWorldOver(t *testing.T, store *memStore, forge *fakeForge, now time.Time w.op.auto = true w.h.Operator = w.op w.merges, w.source = newFakeMerges(forge), &fakeSource{} - w.h.Merges, w.h.Source = w.merges, w.source + w.old = &fakeOld{states: map[string]map[string]oldEntry{}} + w.h.Merges, w.h.Old, w.h.Source = w.merges, w.old, w.source w.merges.h = w.h w.owed = newFakeOwed(forge) w.owed.h = w.h diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go index 14a0cf246..9ba599e69 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go @@ -35,6 +35,7 @@ type Holder struct { // (approval.go); Merges and Source are the gate's merges and source (merges.go). Operator Operator Merges MergeStore + Old OldState Source SourceStore // Owed says what a transition owes the forge as the seat's event owed (owed.go). Owed OwedSayer diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go index e5af7a18c..8f1721dfd 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go @@ -77,6 +77,7 @@ func main() { h.SaySeat = stdio.SeatPublish h.Records = &Records{Ledger: sdkLedger{}, Now: h.Now, Retention: h.recordRetention, Say: stdio.SeatPublish} h.Merges = recordedMerges{MergeStore: kvMerges{}, records: h.Records, members: h.groupMembers} + h.Old = kvOld{} h.Consumer = &StepConsumer{Records: h.Records, Families: stepFamilies, Now: h.Now, Logf: logf} l := &listening{now: "not yet: starting"} go run(h, l) diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go index 1fd3797f7..1404f6237 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go @@ -28,6 +28,26 @@ type MergeStore interface { Put(group string, m Merge) error } +// OldState reads the gate's old state owed, which the switch says again as switched acts (switch.go): each key, and one +// key's value with its revision. +type OldState interface { + Keys(state string) ([]string, error) + Get(state, key string) (value []byte, revision uint64, err error) +} + +// kvOld is the old state over the runtime. +type kvOld struct{} + +func (kvOld) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } + +func (kvOld) Get(state, key string) ([]byte, uint64, error) { + e, err := stdio.State(state).Get(key) + if err != nil || e == nil { + return nil, 0, err + } + return e.Value, e.Revision, nil +} + // SourceStore is the gate's bucket source, one key, moved by compare-and-set. type SourceStore interface { // Move reads source.main and writes what update makes of it, unless update says nothing is to be written; a write diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go index e9fb985e6..2a04a3214 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go @@ -541,15 +541,36 @@ func TestTheSwitchCarriesTheGroupsInFlightIntoRecordsOnce(t *testing.T) { l := newLedgerFake(t) w.h.Records = &Records{Ledger: l, Now: w.h.Now} w.h.Merges = recordedMerges{MergeStore: w.merges, records: w.h.Records, members: w.h.groupMembers} + // W11.17d: what the old state owed is said as switched acts: every key of owed (G §169). + statusKey, _ := mdc.OwedKey("novox/app", 7, readyHead, statusAct("mesh/delivery")) + w.old.put(mdc.SwitchedOwed, statusKey, []byte(`{"act":"status","context":"mesh/delivery","state":"success","text":"ready","target":"none"}`), 4) + switched := func() []meshdelivery.Owed { + var out []meshdelivery.Owed + for _, o := range w.owed.said { + if _, _, _, ok := mdc.ParseSwitchedActID(o.ID); ok { + out = append(out, o) + } + } + return out + } opened, found, err := w.h.Switch() if err != nil || opened != 1 || found != 0 { t.Fatalf("the first switch: %d opened, %d found, %v", opened, found, err) } + carried := switched() + wantStatus, _ := mdc.SwitchedActID(mdc.SwitchedOwed, statusKey, 4) + if len(carried) != 1 || carried[0].ID != wantStatus || carried[0].Act != meshdelivery.OwedActStatus || carried[0].Context != "mesh/delivery" || + carried[0].Commit != readyHead { + t.Fatalf("the switch carried %+v", carried) + } asked := w.op.count() opened, found, err = w.h.Switch() if err != nil || opened != 0 || found != 1 || w.op.count() != asked { t.Fatalf("the switch again: %d opened, %d found, %v, %d asks (was %d)", opened, found, err, w.op.count(), asked) } + if again := switched(); len(again) != 2 || again[1].ID != wantStatus { + t.Fatalf("the switch again said %+v, wanted the same act under the same id", again) + } w.op.answer(w.op.open()[0], OptionApprove) w.settleAll() w.settleAll() diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/switch.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/switch.go index 02f9893bc..8b4c14264 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/switch.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/switch.go @@ -8,6 +8,8 @@ package main // second record and asks nothing again. import ( + "encoding/json" + "errors" "fmt" "sort" @@ -123,9 +125,67 @@ func (h *Holder) Switch() (opened, found int, err error) { } } } + if err := h.carryOwed(); err != nil { + return opened, found, err + } return opened, found, nil } +// oldMerge is a group's merge as the old state merges holds it, with its key and revision; nil when it holds none. +func (h *Holder) oldMerge(group string) (*Merge, uint64, error) { + if h.Old == nil { + return nil, 0, nil + } + key, err := mdc.MergesKey(group) + if err != nil { + return nil, 0, err + } + raw, rev, err := h.Old.Get(mdc.SwitchedMerges, key) + if err != nil || raw == nil { + return nil, 0, err + } + m, err := DecodeMerge(raw) + if err != nil { + return nil, 0, err + } + return &m, rev, nil +} + +// carryOwed says every act the old state owed holds as owed, under its switched id (the state, the key and its revision: +// meshdeliveryclient.SwitchedActID), so the forge bridge performs what the old bridge did not and counts as performed what +// it did (design 47 G §169). No merge act is made from the old state merges (G §167, §542). Said at every switch, it is the same act each time. A key that cannot be read as an act is said +// as refused and skipped. +func (h *Holder) carryOwed() error { + if h.Old == nil || h.Owed == nil { + return nil + } + now := h.Now().UTC() + keys, err := h.Old.Keys(mdc.SwitchedOwed) + if err != nil { + return fmt.Errorf("the state owed cannot be read for the switch: %w", err) + } + for _, key := range keys { + raw, rev, err := h.Old.Get(mdc.SwitchedOwed, key) + if err != nil { + return fmt.Errorf("the owed act %s cannot be read for the switch: %w", key, err) + } + if raw == nil { + continue + } + repository, number, head, _, kerr := mdc.ParseOwedKey(key) + var o Owed + id, ierr := mdc.SwitchedActID(mdc.SwitchedOwed, key, rev) + if derr := json.Unmarshal(raw, &o); kerr != nil || derr != nil || ierr != nil { + h.Refuse(fmt.Errorf("the owed act %s is not carried by the switch: %v", key, errors.Join(kerr, derr, ierr))) + continue + } + if err := h.Owed.Say(seatOwedOf(o, id, repository, number, head, &now)); err != nil { + return fmt.Errorf("the owed act %s cannot be said for the switch: %w", key, err) + } + } + return nil +} + // groupMembers are the members' ids of the milestone a merge is handed under, under the lock. func (h *Holder) groupMembers(name string) []string { h.mu.Lock()