jschoubben
06cf3c04e5
The outbound half went behind `Bus` and the transport stopped reaching its callers; this is the other half, and the larger one. Every handler took `amqp.Delivery`, so the serving loop could not move to another bus without moving enrolment, reports, builds, upgrades and catch-up with it in one breath. `Control` states one message in the mesh's words — took it, dropped it, or held it for the store — and `Inbound` is where messages come from. The AMQP implementation is today's loop moved rather than changed: same queues, same prefetch, same holding, because the mesh is running on it and a bus nothing speaks yet is no reason to alter the one every node is on. The window (window.go) is now what decides, instead of the conditions that were inlined in the loop. Two things that surfaced in the wiring: **Supersession is asked before the store, not after.** A report about a declaration the mesh has moved past would otherwise wait out a restarting store to be written and then overwrite what the node is doing now. **Half of a report is not about a declaration, and that half is never stale.** What the machine *is* — the tunnel it took over, the ports its own bundle holds, what an adopted node found, a node moving its overlay key — reaches the mesh on a report and nowhere else. A rekey set aside as stale is a node whose overlay key never moves, and no retry is coming, because the node said it once. So staleness is asked only of a report that is purely an apply's account. The one thing holding-in-memory can do that holding-in-the-server cannot is named rather than hidden: `About` sets aside a held message when a newer one about the same thing arrives, and the bus being built ignores it because the digest answers the same question.
84 lines
3.3 KiB
Go
84 lines
3.3 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
type heardWith struct{ err error }
|
|
|
|
func (h heardWith) Heard(context.Context, Report) error { return h.err }
|
|
|
|
// switchable answers with whatever it is set to — the store away, then back.
|
|
type switchable struct{ err error }
|
|
|
|
func (h *switchable) Heard(context.Context, Report) error { return h.err }
|
|
|
|
func aReport(node, declared string) Report {
|
|
return Report{Node: node, Declared: declared, Applied: []string{"store"}}
|
|
}
|
|
|
|
// A report the store could not take right now is held, unsettled, and recorded when the store is
|
|
// back; one the store answered no to is acknowledged; one recorded is acknowledged (issue 082, 083).
|
|
func TestAReportTheStoreCouldNotTakeIsHeldAndOneItRefusedIsNot(t *testing.T) {
|
|
store := &switchable{err: errors.Join(ErrTryAgain, errors.New("starting up"))}
|
|
s, in := serving()
|
|
s.listener = store
|
|
held := &settled{}
|
|
s.act(context.Background(), in.sends(t, held, KindReport, aReport("anchor", "d1")))
|
|
if !held.unsettled() || len(in.held) != 1 {
|
|
t.Fatalf("a report the store could not take was not held: %+v, %d held", held, len(in.held))
|
|
}
|
|
store.err = nil
|
|
in.retries(context.Background(), s)
|
|
if !held.acked || len(in.held) != 0 {
|
|
t.Fatalf("a held report was not recorded once the store was back: %+v, %d held", held, len(in.held))
|
|
}
|
|
|
|
refused := &settled{}
|
|
s, in = serving()
|
|
s.listener = heardWith{err: errors.New("a report named no node")}
|
|
s.act(context.Background(), in.sends(t, refused, KindReport, aReport("anchor", "d1")))
|
|
if !refused.acked || refused.nacked {
|
|
t.Fatalf("a report the store answered no to was not acknowledged: %+v", refused)
|
|
}
|
|
}
|
|
|
|
// A newer report from the same node supersedes one of its reports still held: recorded after the
|
|
// newer, the older would overwrite what the node is doing now.
|
|
func TestANewerReportSupersedesAHeldOneFromTheSameNode(t *testing.T) {
|
|
s, in := serving()
|
|
s.listener = heardWith{err: errors.Join(ErrTryAgain, errors.New("starting up"))}
|
|
older, newer, other := &settled{}, &settled{}, &settled{}
|
|
s.act(context.Background(), in.sends(t, older, KindReport, aReport("anchor", "d1")))
|
|
s.act(context.Background(), in.sends(t, other, KindReport, aReport("laptop", "d7")))
|
|
s.act(context.Background(), in.sends(t, newer, KindReport, aReport("anchor", "d2")))
|
|
if !older.acked {
|
|
t.Fatalf("the older report was not set aside by the newer: %+v", older)
|
|
}
|
|
if !newer.unsettled() || !other.unsettled() || len(in.held) != 2 {
|
|
t.Fatalf("the newer report and another node's were not both held: newer %+v other %+v, %d held",
|
|
newer, other, len(in.held))
|
|
}
|
|
}
|
|
|
|
// A store that has not come back within the bound is not restarting: the report is let go, loudly,
|
|
// rather than held for ever.
|
|
func TestAReportIsLetGoOnceTheStoreHasBeenGoneTooLong(t *testing.T) {
|
|
s, in := serving()
|
|
s.listener = heardWith{err: errors.Join(ErrTryAgain, errors.New("connection refused"))}
|
|
s.giveUp = time.Millisecond
|
|
held := &settled{}
|
|
s.act(context.Background(), in.sends(t, held, KindReport, aReport("anchor", "d1")))
|
|
if !held.unsettled() {
|
|
t.Fatalf("the first failure was not held: %+v", held)
|
|
}
|
|
time.Sleep(5 * time.Millisecond)
|
|
in.retries(context.Background(), s)
|
|
if !held.acked || len(in.held) != 0 {
|
|
t.Fatalf("a report past the bound was not let go: %+v, %d held", held, len(in.held))
|
|
}
|
|
}
|