diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/bus_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/bus_test.go index 3a40582f9..bcbf1772a 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/bus_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/bus_test.go @@ -15,6 +15,7 @@ import ( "io" "os" "os/exec" + "reflect" "strings" "sync" "testing" @@ -56,6 +57,8 @@ type testBus struct { // the teller's answers, as its tool teller_answers says them, and the delivery view's reach as view_reach says it. mu sync.Mutex answers map[string][]byte + // the gate's ledger `records`, as the tool runner keeps it: it outlives one run of the gate, as the bus's does. + ledger *ledgerFake } func newTestBus(t *testing.T) *testBus { @@ -119,8 +122,6 @@ type runtime struct { next int wait map[string]chan json.RawMessage - // the gate's ledger, as the tool runner keeps it. - ledger *ledgerFake // what the bundle subscribed to, and asked of the mesh. subs []string asked []askedOfMesh @@ -406,24 +407,24 @@ func (r *runtime) answer(m rpcMessage) { switch m.Method { case buscalls.LedgerApply, buscalls.LedgerGet: // The gate's ledger `records` (W11.1), as the tool runner keeps it: the record test's ledger in memory. - r.mu.Lock() - defer r.mu.Unlock() - if r.ledger == nil { - r.ledger = newLedgerFake(r.t) + r.b.mu.Lock() + defer r.b.mu.Unlock() + if r.b.ledger == nil { + r.b.ledger = newLedgerFake(r.t) } if m.Method == buscalls.LedgerGet { var in buscalls.InstanceRequest if err := json.Unmarshal(m.Params, &in); err != nil { return nil, err } - state, seq, err := r.ledger.Get(in.Instance) + state, seq, err := r.b.ledger.Get(in.Instance) return buscalls.InstanceState{Instance: in.Instance, State: state, Sequence: seq}, err } var in buscalls.ApplyRequest if err := json.Unmarshal(m.Params, &in); err != nil { return nil, err } - seq, err := r.ledger.Apply(in.Instance, in.Operation, in.Body) + seq, err := r.b.ledger.Apply(in.Instance, in.Operation, in.Body) return buscalls.Applied{Taken: err == nil, Sequence: seq}, err } kv := r.b.kvs[p.State] @@ -656,6 +657,49 @@ func TestAPassingVerdictAsksTheApproveAndItsApproveMergesWithinSecondsOverARealB return err == nil && json.Unmarshal(e.Value(), &d) == nil && d.State == Merged }) rt.noControllerAsk() + // W11.16 (acceptance 34, 39): the delivery is traced in one record of the ledger, which says its fill; the gate is stopped + // and started again over the same bus and ledger: the record is whole and opened once, the merge is not handed again, and + // the ask of the Approve already given is not asked again. + recordsNow := func(r *runtime) map[string]any { + var said struct { + Result struct { + Content []struct{ Text string } `json:"content"` + } `json:"result"` + } + raw := r.call(t, "gate_records", map[string]any{}) + if err := json.Unmarshal([]byte(raw), &said); err != nil || len(said.Result.Content) == 0 { + t.Fatalf("gate_records answered %s", raw) + } + var out map[string]any + if err := json.Unmarshal([]byte(said.Result.Content[0].Text), &out); err != nil { + t.Fatalf("gate_records says %s", said.Result.Content[0].Text) + } + return out + } + before := recordsNow(rt) + if kept, _ := before["kept"].(float64); kept != 1 || before["fill"] == nil || before["fill"] == "" { + t.Fatalf("one delivery is traced in %v records, fill %v", before["kept"], before["fill"]) + } + merges := len(rt.mergesSaid()) + _ = rt.in.Close() + _ = rt.cmd.Wait() + rt2 := b.start(t) + evctx2, cancel2 := context.WithCancel(ctx) + defer cancel2() + rt2.deliverEvents(evctx2) + rt2.untilListening() + after := recordsNow(rt2) + if !reflect.DeepEqual(before["records"], after["records"]) || after["kept"] != before["kept"] { + t.Fatalf("the record changed over a restart: %v then %v", before["records"], after["records"]) + } + rt2.until("the restarted gate to hold the delivery merged", eventWithin, func() bool { + e, err := b.kvs["deliveries"].Get(ctx, kvKey(IDOf("novox/app", head))) + var d Delivery + return err == nil && json.Unmarshal(e.Value(), &d) == nil && d.State == Merged + }) + if n := len(rt2.mergesSaid()); n != 0 || len(onlyAskKeys(b)) > 1 { + t.Fatalf("the restarted gate said the merge %d more times (%d before), or asked again: %v", n, merges, onlyAskKeys(b)) + } // Quiet follows: nothing asks the forge or the controller again with no event to say so. quiet := len(rt.asked) time.Sleep(1500 * time.Millisecond) @@ -667,6 +711,12 @@ func TestAPassingVerdictAsksTheApproveAndItsApproveMergesWithinSecondsOverARealB } } +// onlyAskKeys are the keys of every ask in the gate's bucket asks. +func onlyAskKeys(b *testBus) []string { + keys, _ := b.kvs["asks"].Keys(context.Background()) + return keys +} + // onlyAsk is the key of the one ask in the gate's bucket asks, or empty. func onlyAsk(b *testBus) string { keys, err := b.kvs["asks"].Keys(context.Background())