novox/mesh-controller
3
0
Fork 0
Code Issues Pull Requests 4 Releases Wiki Activity

One bus: the AMQP transport is gone from the controller

Browse Source 
The mesh runs on the seat's bus alone (novox/hq ADR 0131, design 28 task 5.5). The old
transport's consume loop, build request, tool ask, management API and account scoping are
deleted, and the bus switch with them; the controller connects to the broker seat and to
nothing else. The store-window tests keep their assertions on a bus-less fake, and the tests
that only made sense for the old transport's in-memory holding go with it.
This commit is contained in:
jschoubben
2026-09-28 03:36:16 +02:00
parent 81e76fa485
commit aecac5bda2
31 changed files with 214 additions and 1915 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/link/ask.go
+22 -60
View File
@@ -3,16 +3,13 @@ package link
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
amqp "github.com/rabbitmq/amqp091-go"
"github.com/nats-io/nats.go"
)
// RPCExchange is where a module's tools are asked over the broker, keyed `<module>.<tool>`, and
// where the answer comes back, keyed by the asker's reply queue (novox/hq ADR 0047).
const RPCExchange = "mesh.rpc"
// Answer is what a module's tool replies: one of the two, never both.
type Answer struct {
Result json.RawMessage `json:"result,omitempty"`
@@ -27,70 +24,35 @@ type Answer struct {
// grants, and should not. The control plane already holds a connection that may, so a person or
// an agent asks through it, and every question passes one process where an audit belongs.
//
// The reply queue is the caller's own, server-named and exclusive, bound to the RPC exchange under
// its own name: a serving module answers through that exchange and never the default one, whose
// permission is per exchange rather than per queue. The correlation is checked rather than
// assumed, as every RPC here is.
func Ask(ctx context.Context, channel *amqp.Channel, module, tool string, args json.RawMessage,
// The answer comes back on the asker's own inbox, which only the asker may read; the serving
// module answers there and nowhere else. The bus refuses a request nothing serves at once, so a
// module that is down or a tool that does not exist is said now rather than after the whole wait.
func Ask(ctx context.Context, bus Bus, module, tool string, args json.RawMessage,
timeout time.Duration) (Answer, error) {
if len(args) == 0 {
args = json.RawMessage(`{}`)
}
replies, err := channel.QueueDeclare("", false, true, true, false, nil)
if err != nil {
return Answer{}, err
}
if err := channel.QueueBind(replies.Name, replies.Name, RPCExchange, false, nil); err != nil {
return Answer{}, fmt.Errorf("cannot bind a reply queue to %s: %w", RPCExchange, err)
}
answers, err := channel.ConsumeWithContext(ctx, replies.Name, "", true, true, false, false, nil)
if err != nil {
return Answer{}, err
}
// Mandatory, so a request nothing consumes comes straight back: a module that is down, or a
// tool that does not exist, is said at once rather than after the whole wait.
returned := channel.NotifyReturn(make(chan amqp.Return, 1))
id := fmt.Sprintf("ask-%d", time.Now().UnixNano())
key := module + "." + tool
if err := channel.PublishWithContext(ctx, RPCExchange, key, true, false, amqp.Publishing{
ContentType: "application/json",
CorrelationId: id,
ReplyTo: replies.Name,
Body: args,
}); err != nil {
return Answer{}, fmt.Errorf("cannot ask %s: %w", key, err)
}
waiting, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
for {
select {
case back := <-returned:
if back.CorrelationId == id {
return Answer{}, fmt.Errorf(
"nothing serves %s: no runtime has bound %q on the broker. The module is not "+
"assigned, its runtime is not up, or it serves no such tool — `status` "+
"says whether the machine carrying it has applied", module, key)
}
case <-waiting.Done():
reply, err := bus.AskTool(ctx, module, tool, args, timeout)
if err != nil {
if errors.Is(err, nats.ErrNoResponders) {
return Answer{}, fmt.Errorf(
"nothing serves %s: no runtime has bound %q on the bus. The module is not "+
"assigned, its runtime is not up, or it serves no such tool — `status` says "+
"whether the machine carrying it has applied", module, key)
}
if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout) {
return Answer{}, fmt.Errorf(
"%s did not answer within %s. Its runtime serves %q when it is up and has bound "+
"the broker — `status` says whether the machine carrying it has applied",
"the bus — `status` says whether the machine carrying it has applied",
module, timeout, key)
case delivery, ok := <-answers:
if !ok {
return Answer{}, fmt.Errorf("the connection closed while waiting for %s", key)
}
if delivery.CorrelationId != id {
continue
}
var answer Answer
if err := json.Unmarshal(delivery.Body, &answer); err != nil {
return Answer{}, fmt.Errorf("%s answered with something unreadable: %w", key, err)
}
return answer, nil
}
return Answer{}, fmt.Errorf("cannot ask %s: %w", key, err)
}
var answer Answer
if err := json.Unmarshal(reply, &answer); err != nil {
return Answer{}, fmt.Errorf("%s answered with something unreadable: %w", key, err)
}
return answer, nil
}
internal/link/build.go
-94
View File
@@ -1,12 +1,7 @@
package link
import (
"context"
"encoding/json"
"fmt"
"time"
amqp "github.com/rabbitmq/amqp091-go"
)
// Asking a machine to build a module, and hearing what came out.
@@ -22,21 +17,6 @@ import (
// holds no opinion about what they contain, and a host that also built things would be a host
// with a container runtime requirement and a git dependency (novox/hq ADR 0005).
// BuildQueue is where build requests wait. One queue, so several build machines can share the
// work and each request is done exactly once — which is what a queue is for and what a
// per-machine routing key would not give.
const BuildQueue = "builds"
// KeyBuilt is what a builder publishes when it has finished, successfully or not.
const KeyBuilt = "built"
// ReplyQueue is where the answer to one request goes.
//
// **Named here rather than left to the broker**, so it can be scoped and reasoned about. A broker
// generates its own name for an unnamed queue, and a builder permitted to write to whatever that
// convention happens to produce works on one broker and silently cannot answer on another.
func ReplyQueue(id string) string { return BuildQueue + ".reply." + id }
// BuildRequest is one module to build.
type BuildRequest struct {
// ID correlates the answer with the asking. Not the module name: two builds of one module can
@@ -118,77 +98,3 @@ type MadeArtifact struct {
Kind string `json:"kind"`
Reference string `json:"reference"`
}
// RequestBuild asks for a module to be built and waits for the answer.
//
// Waiting rather than returning immediately, because the thing a person wants after asking for a
// build is to know whether it worked. A build that is dispatched and forgotten needs somewhere to
// look afterwards, and there is nowhere yet.
func RequestBuild(ctx context.Context, channel *amqp.Channel, request BuildRequest,
timeout time.Duration) (BuildResult, error) {
// Its own queue for the answer, declared before the ask. Consuming from the shared exchange
// would mean competing with the control plane's own consumer for a message meant for this
// caller — which is the fault this package's own doc comment records having had.
replies, err := channel.QueueDeclare(ReplyQueue(request.ID), false, true, true, false, nil)
if err != nil {
return BuildResult{}, err
}
// Bound to the exchange, and the answer comes back through it.
//
// **A builder never publishes to the default exchange**, because permission there is per
// exchange and not per queue — a builder allowed to use it could publish into any node's
// queue, which is the privilege a build machine most obviously should not have. Found by
// running it: the builder built, could not answer, and the connection closed saying only
// "not allowed to publish to exchange ''".
//
// The cost is that every asker sees every result, which is why the correlation is checked
// below rather than assumed.
if err := channel.QueueBind(replies.Name, KeyBuilt, Exchange, false, nil); err != nil {
return BuildResult{}, err
}
answers, err := channel.ConsumeWithContext(ctx, replies.Name, "", true, true, false, false, nil)
if err != nil {
return BuildResult{}, err
}
body, err := json.Marshal(request)
if err != nil {
return BuildResult{}, err
}
if err := channel.PublishWithContext(ctx, "", BuildQueue, false, false, amqp.Publishing{
ContentType: "application/json",
DeliveryMode: amqp.Persistent,
CorrelationId: request.ID,
ReplyTo: replies.Name,
Body: body,
}); err != nil {
return BuildResult{}, err
}
waiting, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
for {
select {
case <-waiting.Done():
return BuildResult{}, fmt.Errorf(
"no builder answered within %s. Something must be consuming %q, and nothing is "+
"— or it is building something that takes longer than this",
timeout, BuildQueue)
case delivery, ok := <-answers:
if !ok {
return BuildResult{}, fmt.Errorf("the connection closed while waiting for a build")
}
var result BuildResult
if err := json.Unmarshal(delivery.Body, &result); err != nil {
return BuildResult{}, fmt.Errorf("a builder answered with something unreadable: %w", err)
}
if result.ID != request.ID {
// Somebody else's answer on this queue. Ignored rather than returned, because
// returning it would attribute one build's outcome to another's.
continue
}
return result, nil
}
}
}
internal/link/builds_current.go
-202
View File
@@ -1,202 +0,0 @@
package link
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
amqp "github.com/rabbitmq/amqp091-go"
)
// The build flow on the bus the mesh runs on today.
//
// Moved behind the seam rather than changed. The queue, the reply binding and the correlation are
// what they were, because the mesh is running on this.
// currentBuilds asks for builds over a channel.
type currentBuilds struct{ channel *amqp.Channel }
// BuildsOverCurrent is the asking side on the bus the mesh has.
func BuildsOverCurrent(channel *amqp.Channel) Builders { return currentBuilds{channel: channel} }
func (b currentBuilds) Close() {}
func (b currentBuilds) Submit(ctx context.Context, request BuildRequest,
wait time.Duration) (BuildResult, error) {
// Its own queue for the answer, declared before the ask. Consuming from the shared exchange
// would mean competing with the controller's own consumer for a message meant for this caller.
replies, err := b.channel.QueueDeclare(ReplyQueue(request.ID), false, true, true, false, nil)
if err != nil {
return BuildResult{}, err
}
// **A builder never publishes to the default exchange**, because permission there is per
// exchange and not per queue — a builder allowed to use it could publish into any node's queue,
// which is the privilege a build machine most obviously should not have. The cost is that every
// asker sees every result, which is why the correlation is checked below rather than assumed.
if err := b.channel.QueueBind(replies.Name, KeyBuilt, Exchange, false, nil); err != nil {
return BuildResult{}, err
}
answers, err := b.channel.ConsumeWithContext(ctx, replies.Name, "", true, true, false, false, nil)
if err != nil {
return BuildResult{}, err
}
body, err := json.Marshal(request)
if err != nil {
return BuildResult{}, err
}
if err := b.channel.PublishWithContext(ctx, "", BuildQueue, false, false, amqp.Publishing{
ContentType: "application/json",
DeliveryMode: amqp.Persistent,
CorrelationId: request.ID,
ReplyTo: replies.Name,
Body: body,
}); err != nil {
return BuildResult{}, err
}
waiting, cancel := context.WithTimeout(ctx, wait)
defer cancel()
for {
select {
case <-waiting.Done():
return BuildResult{}, waitingFor(wait)
case delivery, ok := <-answers:
if !ok {
return BuildResult{}, errors.New("the connection closed while waiting for a build")
}
result, mine, err := theOutcomeOf(delivery.Body, request.ID)
if err != nil {
return BuildResult{}, err
}
if mine {
return result, nil
}
}
}
}
// --- the machine's side ---------------------------------------------------------------------
type currentMachine struct {
conn *amqp.Connection
channel *amqp.Channel
on string
}
// MachineOverCurrent takes build work over a channel.
func MachineOverCurrent(conn *amqp.Connection, channel *amqp.Channel, on string) BuildMachine {
return &currentMachine{conn: conn, channel: channel, on: on}
}
func (m *currentMachine) Close() {}
func (m *currentMachine) Take(ctx context.Context, do func(context.Context, Build)) error {
if _, err := m.channel.QueueDeclare(BuildQueue, true, false, false, false, nil); err != nil {
return err
}
// One at a time. A machine that took five requests at once would run five container builds
// against one runtime and finish all of them slower than it would have finished the first — and
// the queue is what shares work between machines, so nothing is lost by it.
if err := m.channel.Qos(1, 0, false); err != nil {
return err
}
// Not auto-acknowledged: a request acknowledged on arrival is a build that vanishes if this
// process dies mid-way, with nobody waiting on it ever hearing why.
requests, err := m.channel.ConsumeWithContext(ctx, BuildQueue, "mesh-builder",
false, false, false, false, nil)
if err != nil {
return err
}
for {
select {
case <-ctx.Done():
return nil
case delivery, ok := <-requests:
if !ok {
return errors.New("the broker closed the connection")
}
var request BuildRequest
if err := json.Unmarshal(delivery.Body, &request); err != nil {
// Unreadable: rejected rather than retried, because the next attempt reads the same
// bytes. Nobody waiting hears an answer, which is correct — there was no request.
_ = delivery.Reject(false)
continue
}
do(ctx, &currentBuild{request: request, delivery: delivery, on: m.on, channel: m.channel})
}
}
}
type currentBuild struct {
request BuildRequest
delivery amqp.Delivery
on string
channel *amqp.Channel
}
func (b *currentBuild) Request() BuildRequest { return b.request }
// Announce answers and announces, which on this bus are two publishes to two exchanges.
//
// The reply goes to whoever asked, correlated to their request; the announcement says to the whole
// mesh that a module now exists at a commit (novox/hq ADR 0072). Only a successful build is
// announced: a failed one produced no module version, and announcing one would put something in the
// graph that was never made.
func (b *currentBuild) Announce(ctx context.Context, result BuildResult) error {
body, err := json.Marshal(result)
if err != nil {
return err
}
if err := b.channel.PublishWithContext(ctx, Exchange, KeyBuilt, false, false, amqp.Publishing{
ContentType: "application/json",
CorrelationId: result.ID,
Body: body,
}); err != nil {
return fmt.Errorf("cannot answer a build request: %w", err)
}
if result.Failed != "" || result.Commit == "" {
return nil
}
// **Announced under both names on this bus, for exactly as long as this bus lives.**
//
// A build's outcome belongs to the role now (novox/hq ADR 0121), so a catalogue built from the
// current manifests listens for the role's name. A catalogue that is *already running* listens for
// the module's, because that is what it was told when it was installed. A rename on a live bus
// needs the publisher and the subscriber to change together, and a merge cannot promise that: one
// of them is deployed first, and in that window the graph silently stops being updated — which is
// the failure this whole change was cleaning up after.
//
// So both, and the order stops mattering. The module's own name goes with the bus, in step 5's
// retirement list; nothing has ever run on the bus being built, so there is no legacy name there
// and this doubling has no counterpart.
announced := announcementOf(result)
if err := EmitEvent(ctx, OverCurrent{Channel: b.channel}, KeyModuleBuilt, "builder", b.on,
announced); err != nil {
return err
}
return EmitEvent(ctx, OverCurrent{Channel: b.channel}, KeyRoleBuilt, TheBuildMachine, b.on,
announced)
}
func (b *currentBuild) Done() error { return b.delivery.Ack(false) }
func (b *currentBuild) Hold(time.Duration) error {
// No delayed redelivery on this bus: handed back at once, which is what it has always done.
return b.delivery.Nack(false, true)
}
// announcementOf is what the mesh is told about a finished build. One function, so the two
// transports cannot describe the same build differently.
func announcementOf(result BuildResult) map[string]any {
return map[string]any{
"module": ModuleOf(result.Manifest), "commit": result.Commit,
"repository": result.Repository, "path": result.Path, "ref": result.Ref,
"manifest": json.RawMessage(result.Manifest), "against": result.Against,
"made": result.Made,
}
}
internal/link/bus.go
-46
View File
@@ -7,7 +7,6 @@ import (
"time"
"github.com/nats-io/nats.go"
amqp "github.com/rabbitmq/amqp091-go"
)
// Bus is what the controller needs of the mesh's bus, **in the mesh's own words rather than a
@@ -39,51 +38,6 @@ type Bus interface {
// --- The bus the mesh runs on today -----------------------------------------------------
// OverCurrent is the bus the mesh runs on today, until the rollout.
type OverCurrent struct{ Channel *amqp.Channel }
func (b OverCurrent) PublishEvent(ctx context.Context, key, source, node string, body []byte) error {
id, err := eventID()
if err != nil {
return err
}
return b.Channel.PublishWithContext(ctx, EventsExchange, key, false, false, amqp.Publishing{
ContentType: "application/json",
DeliveryMode: amqp.Persistent,
MessageId: id,
Timestamp: time.Now().UTC(),
Body: body,
Headers: amqp.Table{
"x-event-id": id,
"x-source": source,
"x-node": node,
"x-time": time.Now().UTC().Format(time.RFC3339),
"content-type": "application/json",
},
})
}
// AskTool is implemented over the existing reply-queue machinery in ask.go; this seam does not
// change how it works today.
func (b OverCurrent) AskTool(ctx context.Context, module, tool string, args []byte, timeout time.Duration) ([]byte, error) {
answer, err := Ask(ctx, b.Channel, module, tool, args, timeout)
if err != nil {
return nil, err
}
return answer.Result, nil
}
func (b OverCurrent) PublishDeclaration(ctx context.Context, node string, body []byte) error {
// To the queue directly rather than through an exchange: a declaration is for one node, and
// routing it by name through a shared exchange would mean a binding per node that nothing
// removes when a node is retired.
return b.Channel.PublishWithContext(ctx, "", QueueFor(node), false, false, amqp.Publishing{
ContentType: "application/json",
DeliveryMode: amqp.Persistent,
Body: body,
})
}
// --- NATS, the bus being built ----------------------------------------------------------------
// OverNATS is the bus as a JetStream context.
internal/link/bus_nats.go
+1 -51
View File
@@ -1,66 +1,16 @@
package link
import (
"context"
"fmt"
"time"
"github.com/nats-io/nats.go"
"github.com/novox/mesh-controller/internal/broker"
)
// OverNats is the controller's outbound on the bus being built: the same three acts the other
// transport has, on the subjects the permissions were derived for (design 25). A declaration is a
// JetStream publish into NODES, where the node's own consumer waits for it; an event is announced on
// the subject its name derives to; a tool is asked by request and reply on the module's tool subject.
type OverNats struct{ JS *broker.JetStream }
// declareSubject is where one node's declaration lands — the NODES stream's subject for it, and the
// only subject that node's consumer delivers. The host subscribes exactly this.
func declareSubject(node string) string { return "mesh.node." + node + ".declare" }
func (b OverNats) PublishDeclaration(ctx context.Context, node string, body []byte) error {
publish, cancel := context.WithTimeout(ctx, 15*time.Second)
defer cancel()
if _, err := b.JS.Context().Publish(declareSubject(node), body, nats.Context(publish)); err != nil {
return fmt.Errorf("declaring to %s: %w", node, err)
}
return nil
}
func (b OverNats) PublishEvent(ctx context.Context, key, source, node string, body []byte) error {
// The key is the subject: the controller's own events are named in full, and what a module
// emits is derived before it reaches here. Headers carry the envelope the other transport put
// in message properties (ADR 0042), so a consumer reads who and when without the payload.
msg := nats.NewMsg(key)
msg.Data = body
msg.Header.Set("x-source", source)
msg.Header.Set("x-node", node)
msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
if err := b.JS.Conn().PublishMsg(msg); err != nil {
return fmt.Errorf("announcing %s: %w", key, err)
}
return nil
}
func (b OverNats) AskTool(ctx context.Context, module, tool string, args []byte, timeout time.Duration) ([]byte, error) {
ask, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
reply, err := b.JS.Conn().RequestWithContext(ask, "mesh.mod."+module+".tool."+tool, args)
if err != nil {
return nil, fmt.Errorf("asking %s.%s: %w", module, tool, err)
}
return reply.Data, nil
}
// ConnectNats is Connect for the bus being built: the controller's inbound and outbound over one
// JetStream connection the caller has already raised the streams on. Nothing is declared here —
// the streams and the controller's consumers are asserted by Raise, before anything is served.
func ConnectNats(js *broker.JetStream, enroller Enroller, listener Listener) *Server {
return &Server{
inbound: Nats(js),
bus: OverNats{JS: js},
bus: OverNATS{JS: js.Context(), Conn: js.Conn()},
js: js,
enroller: enroller,
listener: listener,
internal/link/enrolment.go
+3 -18
View File
@@ -24,10 +24,9 @@ import (
// — it holds both grants and asks each for its part, which is what the process running them is
// for.
type Enrolment struct {
Inventory *inventory.Inventory
Identity *identity.Identity
Management *broker.Management
Broker broker.Broker
Inventory *inventory.Inventory
Identity *identity.Identity
Broker broker.Broker
// OnNATS says the mesh's own traffic is on the bus being built, so a node's credential is
// minted into the mesh's records and composed into the bus's user list rather than pushed
@@ -194,17 +193,6 @@ func (e Enrolment) Enrol(ctx context.Context, request EnrolRequest) (reply Enrol
reply.Password = password
}
case e.Management != nil:
password, err := freshPassword()
if err != nil {
return EnrolReply{}, err
}
if err := e.Management.CreateNodeAccount(ctx, node.Name, password); err != nil {
log.Printf("%s is enrolled and its broker password could not be replaced, so it keeps "+
"the token's secret as its password: %v", node.Name, err)
} else {
reply.Password = password
}
}
if profile != nil {
@@ -261,9 +249,6 @@ func freshPassword() (string, error) {
var _ Enroller = Enrolment{}
// ErrNoBrokerManagement is returned when an account cannot be made because nothing was configured.
var ErrNoBrokerManagement = errors.New("no broker management configured")
// Heard records what a node reported about itself.
//
// A node states; the owning context writes (novox/hq ADR 0006). What a node says it applied is
internal/link/receive.go
-13
View File
@@ -78,19 +78,6 @@ type Control interface {
// Took settles the message: acted on, or understood and needing no action.
Took() error
// About names what this message is about — a node's report, one module's move, one build's
// outcome — and is said before the store is asked.
//
// A transport that holds messages **in memory** uses it to set aside anything older it is
// holding about the same thing: the older is the past, and letting it come back after the
// newer was acted on would undo the newer.
//
// **This is the one thing holding-in-memory can do that holding-in-the-server cannot**, and
// naming it here rather than hiding it is deliberate. On the bus being built the message
// belongs to the server and comes back whatever happened meanwhile, so this is ignored and the
// digest a report carries answers the same question instead (window.go, design 25 §3).
About(what string)
// Hold keeps the message and asks for it again after the delay — the store window.
Hold(after time.Duration) error
internal/link/receive_current.go
-321
View File
@@ -1,321 +0,0 @@
package link
import (
"context"
"errors"
"fmt"
"time"
amqp "github.com/rabbitmq/amqp091-go"
)
// The consume side on the bus the mesh runs on today.
//
// Everything here was the serving loop's until the seam went in: the queues, the binds, the
// prefetch, and the list of messages the store could not take yet. It moved rather than changed —
// the behaviour this transport has is the behaviour it had, because the mesh is running on it and
// a bus nothing speaks yet is no reason to alter the one every node is on (ADR 0116).
// Prefetch is how many messages the bus hands the controller before it has settled them.
//
// More than one because a message the store could not take is held, unsettled, while the loop
// goes on answering others — an enrolment above all, which a host is waiting on (novox/hq issue
// 083). Bounded, because what is held is also what the bus has not kept on its own disk as
// pending.
const Prefetch = 64
// PrefetchHeadroom is how much of the prefetch is never held, so the loop always has messages to
// answer — an enrolment above all — while others wait for the store.
const PrefetchHeadroom = 8
// TryAgainAfter is how often the held are looked at. A store comes back in seconds, and a report a
// few seconds late is still current.
const TryAgainAfter = 2 * time.Second
// currentInbound consumes what nodes say over the bus the mesh has.
type currentInbound struct {
conn *amqp.Connection
channel *amqp.Channel
// upgrades and catchups are bound only when something is listening (Also).
upgrades bool
catchups bool
// held is every message the store could not take, by the bus's own delivery tag. Kept here
// rather than in the serving loop because holding a delivery unacknowledged is this
// transport's way of keeping it, and the other's is to hand it back to the server.
held map[uint64]*holding
// again is how often the held are looked at; zero means TryAgainAfter. Set by tests.
again time.Duration
}
// holding is one message kept for the store, and when to try it again.
type holding struct {
message *currentControl
due time.Time
about string
}
// Current is the consume side of the bus the mesh runs on today.
func Current(conn *amqp.Connection, channel *amqp.Channel) Inbound {
return &currentInbound{conn: conn, channel: channel, held: map[uint64]*holding{}}
}
// Also binds the queue one more kind arrives on.
//
// The kinds nodes publish all share one queue and are bound at Connect, because a node may
// publish any of them and binding one while forgetting another is a message the bus accepts, finds
// no queue for, and drops — the publisher sees success and the consumer sees nothing. The two that
// are events get their own queue each, and only when something is listening.
func (c *currentInbound) Also(kind string) error {
switch kind {
case KindSourceMoved:
// Not followed on the bus the mesh is leaving: the forge's merges are announced on the
// new one, and this transport goes with the move (design 28, task 5.5).
return nil
case KindModuleMoved:
if err := c.bindEvent(UpgradeQueue, KeyModuleUpgraded); err != nil {
return err
}
c.upgrades = true
case KindCatchUp:
if err := c.bindEvent(CatchUpQueue, KeyCatchingUp); err != nil {
return err
}
c.catchups = true
default:
return fmt.Errorf("nothing binds a queue for %s on this bus", kind)
}
return nil
}
func (c *currentInbound) bindEvent(queue, key string) error {
if _, err := c.channel.QueueDeclare(queue, true, false, false, false, nil); err != nil {
return fmt.Errorf("cannot declare the %s queue: %w", queue, err)
}
if err := c.channel.QueueBind(queue, key, EventsExchange, false, nil); err != nil {
return fmt.Errorf("cannot bind %s to %s/%s: %w", queue, EventsExchange, key, err)
}
return nil
}
func (c *currentInbound) Close() {}
// Receive consumes until the context ends.
//
// One consumer per queue, deliberately: with two on one queue the bus would round-robin between
// them and each would receive half of what it expects — a fault this project has already had,
// between a module's daemon and its capability server.
func (c *currentInbound) Receive(ctx context.Context, act func(context.Context, Control)) error {
// A bounded prefetch rather than one. The loop still takes messages one at a time; what the
// prefetch buys is that a message the store could not take can be held while the loop goes on
// to the next, instead of every enrolment waiting behind it (novox/hq issue 083). Anything
// held goes back to the bus if the controller stops, because nothing held is acknowledged.
if err := c.channel.Qos(Prefetch, 0, false); err != nil {
return err
}
deliveries, err := c.channel.ConsumeWithContext(ctx, ControlQueue, "control-plane",
false, false, false, false, nil)
if err != nil {
return err
}
// Its own queue and its own consumer for each event, for the reason above: two consumers on
// one queue split its messages between them, and an upgrade or a catch-up request going to
// whichever half was not listening is a gap that looks like a working mesh.
var upgrades, catchups <-chan amqp.Delivery
if c.upgrades {
upgrades, err = c.channel.ConsumeWithContext(ctx, UpgradeQueue, "control-plane-upgrades",
false, false, false, false, nil)
if err != nil {
return err
}
}
if c.catchups {
catchups, err = c.channel.ConsumeWithContext(ctx, CatchUpQueue, "control-plane-catchup",
false, false, false, false, nil)
if err != nil {
return err
}
}
closed := c.conn.NotifyClose(make(chan *amqp.Error, 1))
again := c.again
if again == 0 {
again = TryAgainAfter
}
ticker := time.NewTicker(again)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return nil
case <-ticker.C:
if ctx.Err() != nil {
return nil
}
c.retryHeld(ctx, act)
case delivery, ok := <-catchups:
if !ok {
if catchups != nil {
return errors.New("the bus stopped delivering catch-up requests")
}
continue
}
act(ctx, c.wrap(KindCatchUp, delivery))
case delivery, ok := <-upgrades:
// A nil channel blocks for ever, so this case simply never fires when nothing is
// listening for upgrades. Closed is different, and means the bus stopped.
if !ok {
if upgrades != nil {
return errors.New("the bus stopped delivering upgrades")
}
continue
}
act(ctx, c.wrap(KindModuleMoved, delivery))
case reason := <-closed:
// Said rather than returned quietly. A controller whose bus connection dropped is a
// mesh where nothing can be told anything, and the reason is the first thing anybody
// will want.
return fmt.Errorf("the bus connection closed: %v", reason)
case delivery, ok := <-deliveries:
if !ok {
return errors.New("the bus stopped delivering")
}
kind, known := kindOfKey[delivery.RoutingKey]
if !known {
// Rejected without requeue: a message nothing understands will not be understood
// on the next attempt either, and requeuing it would spin.
_ = delivery.Reject(false)
continue
}
act(ctx, c.wrap(kind, delivery))
}
}
}
// kindOfKey is how this transport's addressing becomes what the mesh calls a message.
var kindOfKey = map[string]string{
KeyEnrol: KindEnrolment,
KeyReport: KindReport,
KeyAlive: KindHeartbeat,
KeyBuilt: KindBuilt,
KeyModuleUpgraded: KindModuleMoved,
KeyCatchingUp: KindCatchUp,
}
func (c *currentInbound) wrap(kind string, delivery amqp.Delivery) *currentControl {
return &currentControl{kind: kind, delivery: delivery, on: c}
}
// retryHeld hands every message whose delay has passed back to the loop. Each handler holds it
// again, settles it, or lets it go past the bound.
func (c *currentInbound) retryHeld(ctx context.Context, act func(context.Context, Control)) {
now := time.Now()
due := make([]*currentControl, 0, len(c.held))
for _, h := range c.held {
if !h.due.After(now) {
due = append(due, h.message)
}
}
for _, m := range due {
if ctx.Err() != nil {
return
}
act(ctx, m)
}
}
// currentControl is one delivery from the bus the mesh has, as the controller reads it.
type currentControl struct {
kind string
delivery amqp.Delivery
on *currentInbound
// about is what this message is about, as the handler named it; empty until it does.
about string
// first is when this message was first held for the store; zero while it has not been.
first time.Time
}
func (m *currentControl) Kind() string { return m.kind }
func (m *currentControl) Body() []byte { return m.delivery.Body }
func (m *currentControl) Redelivered() bool { return m.delivery.Redelivered }
func (m *currentControl) HeldFor() time.Duration {
if m.first.IsZero() {
return 0
}
return time.Since(m.first)
}
// Answer publishes to the reply queue the request named.
func (m *currentControl) Answer(ctx context.Context, body []byte) error {
if m.delivery.ReplyTo == "" {
return errors.New("that request named no reply queue, so nothing can be told the answer")
}
return m.on.channel.PublishWithContext(ctx, "", m.delivery.ReplyTo, false, false,
amqp.Publishing{
ContentType: "application/json",
CorrelationId: m.delivery.CorrelationId,
Body: body,
})
}
func (m *currentControl) Took() error {
m.forget()
return m.delivery.Ack(false)
}
// Drop rejects without requeue: on this bus that is what "understood, and not worth another
// attempt" is spelled as, and it is what feeds a dead-letter queue where one is configured.
func (m *currentControl) Drop() error {
m.forget()
return m.delivery.Reject(false)
}
// About names what this message is about, and lets go of whatever is held about the same thing:
// the held one is the past, and acting on it after this one would undo this one. Acknowledged
// rather than left to come back, because a held message nothing will act on is a place in the
// prefetch nothing gets back.
func (m *currentControl) About(what string) {
m.about = what
if what == "" {
return
}
for tag, h := range m.on.held {
if h.about != what || tag == m.delivery.DeliveryTag {
continue
}
delete(m.on.held, tag)
_ = h.message.delivery.Ack(false)
}
}
// Hold keeps the message unacknowledged and sets it aside to be handed back after the delay.
//
// Held no further than the prefetch leaves room: past that the bus would hand the loop nothing
// new — enrolments included — until something held was let go. A message that cannot be held says
// so, and the handler settles it its own way.
func (m *currentControl) Hold(after time.Duration) error {
if m.on.held == nil {
m.on.held = map[uint64]*holding{}
}
if _, already := m.on.held[m.delivery.DeliveryTag]; !already {
if len(m.on.held) >= Prefetch-PrefetchHeadroom {
return fmt.Errorf("%d messages are already held for the store, and holding more "+
"would stop the queue", len(m.on.held))
}
m.first = time.Now()
}
m.on.held[m.delivery.DeliveryTag] = &holding{
message: m, due: time.Now().Add(after), about: m.about,
}
return nil
}
func (m *currentControl) forget() {
if m.on != nil {
delete(m.on.held, m.delivery.DeliveryTag)
}
}
internal/link/receive_current_test.go
-71
View File
@@ -1,71 +0,0 @@
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 := &currentInbound{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 &currentControl{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)
}
internal/link/receive_fake_test.go
+94
View File
@@ -0,0 +1,94 @@
package link
import (
"context"
"encoding/json"
"testing"
"time"
)
// A harness for the controller's decisions about a message, with no bus behind it.
//
// What these tests exercise is the server's verdict — taken, dropped, held for the store, let go
// once the store has been away too long — and the transport's part in that is only to keep a held
// message and hand it back. A fake that does exactly that stands in for the bus, so what is
// asserted is how a message was settled and the decision needs no server.
// settled is how the bus was told to settle one message.
type settled struct{ acked, nacked, requeued, rejected bool }
// 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 }
// fakeInbound keeps the messages the controller held, the way a stream would.
type fakeInbound struct{ held map[uint64]*fakeControl }
var tag uint64
// 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, *fakeInbound) {
in := &fakeInbound{held: map[uint64]*fakeControl{}}
return &Server{inbound: in, log: quiet()}, in
}
func (c *fakeInbound) Also(string) error { return nil }
func (c *fakeInbound) Close() {}
func (c *fakeInbound) Receive(context.Context, func(context.Context, Control)) error {
return nil
}
// sends is one message arriving, as the controller reads it.
func (c *fakeInbound) 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 &fakeControl{kind: kind, body: body, tag: tag, to: to, on: c}
}
// retries hands every held message back to the controller, the way a stream redelivers.
func (c *fakeInbound) retries(ctx context.Context, s *Server) {
for tag, m := range c.held {
delete(c.held, tag)
m.redelivered = true
s.act(ctx, m)
}
}
type fakeControl struct {
kind string
body []byte
tag uint64
to *settled
on *fakeInbound
redelivered bool
since time.Time
}
func (m *fakeControl) Kind() string { return m.kind }
func (m *fakeControl) Body() []byte { return m.body }
func (m *fakeControl) Redelivered() bool { return m.redelivered }
func (m *fakeControl) About(string) {}
func (m *fakeControl) Answer(context.Context, []byte) error { return nil }
func (m *fakeControl) Took() error { m.to.acked = true; return nil }
func (m *fakeControl) Drop() error { m.to.rejected = true; return nil }
// HeldFor is how long the store has been waited on for this message — zero on a first delivery.
func (m *fakeControl) HeldFor() time.Duration {
if m.since.IsZero() {
return 0
}
return time.Since(m.since)
}
func (m *fakeControl) Hold(time.Duration) error {
if m.since.IsZero() {
m.since = time.Now()
}
m.on.held[m.tag] = m
return nil
}
internal/link/receive_nats.go
+4 -8
View File
@@ -23,6 +23,10 @@ import (
// it first could not take it, and a controller that restarts mid-window has nothing to lose.
// natsInbound consumes what nodes and modules say over NATS.
// Prefetch is how many controls the controller holds unacknowledged at once: the store window
// (ADR 0083) is the server's, and this is what it may hand this process ahead of its acting.
const Prefetch = 64
type natsInbound struct {
js *broker.JetStream
// follows is the kinds asked for beyond what nodes say (Also). The events those are are the
@@ -222,14 +226,6 @@ func (m *natsControl) HeldFor() time.Duration {
return time.Since(first)
}
// About is nothing here, and that is the point.
//
// Setting a held message aside when a newer one about the same thing arrives is what a controller
// holding deliveries in memory can do. A naked message belongs to the server and comes back
// whatever happened meanwhile, so the question "is this the past?" is answered by what the message
// says instead — the digest of the declaration a report is about (window.go, design 25 §3).
func (m *natsControl) About(string) {}
// Answer publishes to the reply subject the request carries **in its payload**.
//
// Not `Respond`, and not the message's reply field: a message a JetStream consumer delivers has had
internal/link/receive_nats_test.go
+4
View File
@@ -4,6 +4,8 @@ import (
"context"
"encoding/json"
"errors"
"io"
"log"
"os"
"sync"
"testing"
@@ -120,6 +122,8 @@ func eventually(t *testing.T, what string, is func() bool) {
// A report published by a node reaches the controller, is recorded, and is acknowledged — so the
// stream does not hold it. A work queue is the check: what is acknowledged leaves it.
func quiet() *log.Logger { return log.New(io.Discard, "", 0) }
func TestNatsAReportIsHeardAndLeavesTheStream(t *testing.T) {
js := aBus(t)
store := &counted{}
internal/link/report_retry_test.go
-18
View File
@@ -46,24 +46,6 @@ func TestAReportTheStoreCouldNotTakeIsHeldAndOneItRefusedIsNot(t *testing.T) {
}
}
// 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) {
internal/link/serve.go
+4 -84
View File
@@ -11,10 +11,6 @@ import (
"log"
"os"
"time"
amqp "github.com/rabbitmq/amqp091-go"
"github.com/novox/mesh-controller/internal/envfile"
)
// AMQPVar is the controller's own connection to the bus the mesh runs on today.
@@ -71,8 +67,6 @@ type Upgrader interface {
type Server struct {
inbound Inbound
bus Bus
conn *amqp.Connection
channel *amqp.Channel
js *broker.JetStream
enroller Enroller
@@ -127,85 +121,18 @@ func (s *Server) Answers(r Replayer) error {
// On the port MESH_BROKER_AMQP_PORT names when the node's settings moved the broker (novox/hq
// 04-ISSUES/102) — the URL is genesis's, sealed, and its port is the one thing in it the node may
// have moved since.
func Connect(enroller Enroller, listener Listener) (*Server, error) {
url, err := envfile.Placed(AMQPVar)
if err != nil {
return nil, err
}
if url == "" {
return nil, fmt.Errorf(
"this control plane has no %s, so it cannot reach its broker. Nodes talk to it over "+
"the broker and nowhere else, so without this it can hold records and answer "+
"nothing", AMQPVar)
}
conn, err := amqp.Dial(url)
if err != nil {
// Not quoted back: the URL carries the controller's own bus password.
return nil, fmt.Errorf("cannot reach the broker named in %s: %w", AMQPVar, err)
}
channel, err := conn.Channel()
if err != nil {
conn.Close()
return nil, err
}
// Declared here rather than assumed. The controller is the only thing that may create them — a
// node's account can write to this exchange and read its own queue, and configure nothing
// else, so a node arriving before the controller has ever run finds nothing and says so,
// rather than quietly creating a topology nobody designed.
if err := channel.ExchangeDeclare(Exchange, "direct", true, false, false, false, nil); err != nil {
conn.Close()
return nil, fmt.Errorf("cannot declare the %s exchange: %w", Exchange, err)
}
// The events exchange too. The controller is not the only publisher on it — modules announce
// onto it with their own accounts — but it is the only thing permitted to create it, for the
// same reason it is the only thing permitted to create the direct one.
if err := channel.ExchangeDeclare(EventsExchange, "topic", true, false, false, false, nil); err != nil {
conn.Close()
return nil, fmt.Errorf("cannot declare the %s exchange: %w", EventsExchange, err)
}
if _, err := channel.QueueDeclare(ControlQueue, true, false, false, false, nil); err != nil {
conn.Close()
return nil, fmt.Errorf("cannot declare the %s queue: %w", ControlQueue, err)
}
// Every key a node may publish. Binding one and forgetting another is a message the bus
// accepts, finds no queue for, and drops — the publisher sees success and the consumer sees
// nothing. That is exactly what happened to reports: `report` was left unbound while `enrol`
// worked, so nodes announced what they had applied into a void for an afternoon.
for _, key := range []string{KeyEnrol, KeyReport, KeyAlive, KeyBuilt} {
if err := channel.QueueBind(ControlQueue, key, Exchange, false, nil); err != nil {
conn.Close()
return nil, fmt.Errorf("cannot bind %s to %s/%s: %w", ControlQueue, Exchange, key, err)
}
}
return &Server{
inbound: Current(conn, channel),
bus: OverCurrent{Channel: channel},
conn: conn,
channel: channel,
enroller: enroller,
listener: listener,
log: newLog(),
}, nil
// Connect is ConnectNats: the mesh has one bus (novox/hq ADR 0131, design 28 task 5.5). Kept as
// the name callers know; the transport it opened before is gone with the bus it spoke to.
func Connect(js *broker.JetStream, enroller Enroller, listener Listener) *Server {
return ConnectNats(js, enroller, listener)
}
func newLog() *log.Logger { return log.New(os.Stdout, "", log.LstdFlags) }
// Channel is the controller's channel, for the command line's own publishing.
func (s *Server) Channel() *amqp.Channel { return s.channel }
func (s *Server) Close() {
if s.inbound != nil {
s.inbound.Close()
}
if s.channel != nil {
_ = s.channel.Close()
}
if s.conn != nil {
_ = s.conn.Close()
}
if s.js != nil {
s.js.Close()
}
@@ -349,7 +276,6 @@ func (s *Server) reported(ctx context.Context, m Control) {
_ = m.Drop()
return
}
m.About("report " + report.Node)
what := fmt.Sprintf("%s's report of declaration %s", report.Node, report.Declared)
if s.listener != nil {
@@ -482,9 +408,6 @@ func (s *Server) wasBuilt(ctx context.Context, m Control) {
_ = m.Drop()
return
}
// Each build result its own subject: none supersedes another, and recording one twice is
// harmless — the build is kept by its id.
m.About("build " + digest(m.Body()))
err := s.recorder.Built(ctx, result)
switch s.decide(ctx, m, fmt.Sprintf("a build result from %s", result.On), "", "", err) {
case Hold:
@@ -516,7 +439,6 @@ func (s *Server) wasBuilt(ctx context.Context, m Control) {
// the catalogue next restarts (issue 083). One request stands for all: a newer one supersedes one
// still held.
func (s *Server) catchingUp(ctx context.Context, m Control) {
m.About("catch-up")
if s.replayer == nil {
s.log.Printf("a catalogue asked to catch up and this control plane has nothing to replay")
_ = m.Took()
@@ -569,7 +491,6 @@ func (s *Server) moved(ctx context.Context, m Control) {
_ = m.Took()
return
}
m.About("upgrade " + u.Module)
if u.Module == "" {
s.log.Printf("an upgrade announcement named no module; ignored")
_ = m.Took()
@@ -620,7 +541,6 @@ func (s *Server) sourceMoved(ctx context.Context, m Control) {
_ = m.Took()
return
}
m.About("merge " + moved.Owner + "/" + moved.Repo + " into " + moved.Base)
if moved.Commit == "" || moved.Repo == "" {
s.log.Printf("a merge announcement named no repository or no commit; ignored")
_ = m.Took()
internal/link/store_window_test.go
+2 -50
View File
@@ -3,7 +3,6 @@ package link
import (
"context"
"errors"
"fmt"
"testing"
"github.com/jackc/pgx/v5/pgconn"
@@ -18,7 +17,8 @@ func (r recordsWith) Built(context.Context, BuildResult) error { return r.err }
type upgradesWith struct{ err error }
func (u upgradesWith) Upgraded(context.Context, Upgraded) error { return u.err }
func (u upgradesWith) Upgraded(context.Context, Upgraded) error { return u.err }
func (u upgradesWith) SourceMoved(context.Context, SourceMoved) error { return u.err }
type replaysWith struct{ err error }
@@ -106,51 +106,3 @@ func TestAMessageHandledDuringShutdownIsLeftForTheBus(t *testing.T) {
t.Fatalf("a build result handled during shutdown was settled, and so lost: %+v", *to)
}
}
// Two identical build results: the newer sets the older aside rather than leaving it unsettled
// for ever, holding a place in the prefetch.
func TestAnIdenticalBuildResultSetsTheHeldOneAside(t *testing.T) {
s, in := serving()
s.recorder = recordsWith{err: restarting}
built := BuildResult{On: "anchor", Repository: "/r", Commit: "abc"}
first, second := &settled{}, &settled{}
s.act(context.Background(), in.sends(t, first, KindBuilt, built))
s.act(context.Background(), in.sends(t, second, KindBuilt, built))
if !first.acked || !second.unsettled() || len(in.held) != 1 {
t.Fatalf("an identical build result did not set the held one aside: first %+v second %+v, %d held",
*first, *second, len(in.held))
}
}
// What is held stops short of the prefetch, so the loop always has room to answer an enrolment.
func TestWhatIsHeldLeavesRoomInThePrefetch(t *testing.T) {
s, in := serving()
s.recorder = recordsWith{err: restarting}
var last *settled
for i := 0; i < Prefetch; i++ {
last = &settled{}
s.act(context.Background(), in.sends(t, last, KindBuilt, BuildResult{On: "anchor", Commit: fmt.Sprint(i)}))
}
if len(in.held) != Prefetch-PrefetchHeadroom {
t.Fatalf("%d messages were held; the ceiling is %d", len(in.held), Prefetch-PrefetchHeadroom)
}
if last.unsettled() {
t.Fatalf("a message past the ceiling was held: %+v", *last)
}
}
// An upgrade handled during shutdown is left for the bus too — the upgrader's error is the
// cancelled context, which is no answer about the announcement.
func TestAnUpgradeHandledDuringShutdownIsLeftForTheBus(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
cancel()
s, in := serving()
s.upgrader = upgradesWith{err: context.Canceled}
to := &settled{}
s.act(ctx, in.sends(t, to, KindModuleMoved, Upgraded{Module: "gitea", Commit: "abcdef0123"}))
if !to.unsettled() {
t.Fatalf("an upgrade was settled during shutdown, and so lost: %+v", *to)
}
}
func (u upgradesWith) SourceMoved(context.Context, SourceMoved) error { return nil }