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.
72 lines
2.3 KiB
Go
72 lines
2.3 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"io"
|
|
"log"
|
|
"testing"
|
|
"time"
|
|
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
)
|
|
|
|
// The harness for the consume side on the bus the mesh runs on today.
|
|
//
|
|
// Messages arrive through the seam, so what these tests exercise is the controller's decision
|
|
// about a message and this transport's way of keeping one — which is what the seam separated. A
|
|
// fake acknowledger stands in for the bus, because what is asserted is how a message was settled
|
|
// and that needs no server.
|
|
|
|
// settled is how the bus was told to settle one message.
|
|
type settled struct{ acked, nacked, requeued, rejected bool }
|
|
|
|
func (a *settled) Ack(uint64, bool) error { a.acked = true; return nil }
|
|
func (a *settled) Nack(_ uint64, _ bool, requeue bool) error {
|
|
a.nacked, a.requeued = true, requeue
|
|
return nil
|
|
}
|
|
func (a *settled) Reject(uint64, bool) error { a.rejected = true; return nil }
|
|
|
|
// unsettled is a message the controller has neither taken nor let go: it is held, and the bus will
|
|
// hand it to whatever consumes next if the controller stops.
|
|
func (a *settled) unsettled() bool { return !a.acked && !a.nacked && !a.rejected }
|
|
|
|
var tag uint64
|
|
|
|
func quiet() *log.Logger { return log.New(io.Discard, "", 0) }
|
|
|
|
// serving is a controller with nothing but a way of receiving, ready for a listener, a recorder,
|
|
// an upgrader or a replayer to be set on it.
|
|
func serving() (*Server, *currentInbound) {
|
|
in := ¤tInbound{held: map[uint64]*holding{}}
|
|
return &Server{inbound: in, bus: OverCurrent{}, log: quiet()}, in
|
|
}
|
|
|
|
// sends is one message arriving over this transport, as the controller reads it.
|
|
func (c *currentInbound) sends(t *testing.T, to *settled, kind string, v any) Control {
|
|
t.Helper()
|
|
body, err := json.Marshal(v)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
tag++
|
|
return ¤tControl{kind: kind, on: c, delivery: amqp.Delivery{
|
|
Acknowledger: to, Body: body, DeliveryTag: tag,
|
|
}}
|
|
}
|
|
|
|
// dueNow brings every held message forward, so a test need not wait out the backoff a real store
|
|
// restart would be given (RedeliverAfter).
|
|
func (c *currentInbound) dueNow() {
|
|
for _, h := range c.held {
|
|
h.due = time.Now().Add(-time.Second)
|
|
}
|
|
}
|
|
|
|
// retries hands every held message back to the controller, the way the ticker does.
|
|
func (c *currentInbound) retries(ctx context.Context, s *Server) {
|
|
c.dueNow()
|
|
c.retryHeld(ctx, s.act)
|
|
}
|