Builds have a history, and failures are rows like any other

A build result was answered to whoever asked and kept nowhere. So "when
did this last build", "why did it fail" and "which machine built what is
running" had no answer, and a build nobody was waiting for was reported
into the void — which is the same as not reporting it.

Failures are recorded too, and that is the point rather than a detail: a
failed build that leaves no trace is indistinguishable from one nobody
asked for, and the difference is the whole of whether somebody should be
looking at something. A build that never learned what it was building
keeps the repository, because that is what a person goes and looks at.

Recording is idempotent on the correlation id, because a result can
arrive twice — as the answer to whoever asked, and on the exchange when
nobody was. Two rows would show one build as two, and which is real is
not answerable afterwards.

The serving control plane now binds `built` as well, so results from
builds it did not ask for are kept. It refuses them loudly when it has
nowhere to put them rather than dropping them, so the broker's own
counters show something arriving that nothing handles.

`builds [<module>]` reads it: what happened lately across the mesh, or
what has happened to one module — the first asked after something goes
wrong, the second when deciding whether to trust something.

What was published is kept with the build, so a digest traces back to
what made it without holding the manifest twice in a place that can
disagree with the first.

internal/link/serve.go

@@ -33,14 +33,32 @@ type Listener interface {
 	Heard(ctx context.Context, report Report) error
 }
 
+// Recorder keeps what builders say.
+//
+// Separate from Listener because they are different things arriving: a report is a node saying
+// what it did with a declaration, and a build result is a machine saying what came of some work.
+// One interface carrying both would mean an implementation of one having to say something about
+// the other.
+type Recorder interface {
+	Built(ctx context.Context, result BuildResult) error
+}
+
 type Server struct {
 	conn     *amqp.Connection
 	channel  *amqp.Channel
 	enroller Enroller
 	listener Listener
+	recorder Recorder
 	log      *log.Logger
 }
 
+// Records tells the server where to keep build results.
+//
+// Set after Connect rather than passed to it, because a control plane that only publishes — the
+// `build` command, which waits for its own answer — needs a connection and no recorder, and
+// making it supply one would have it construct something it never uses.
+func (s *Server) Records(r Recorder) { s.recorder = r }
+
 // Connect opens the control plane's own connection to the broker.
 func Connect(enroller Enroller, listener Listener) (*Server, error) {
 	url := strings.TrimSpace(os.Getenv(AMQPVar))
@@ -78,7 +96,7 @@ func Connect(enroller Enroller, listener Listener) (*Server, error) {
 	// 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} {
+	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)
@@ -121,7 +139,8 @@ func (s *Server) Serve(ctx context.Context) error {
 	}
 
 	closed := s.conn.NotifyClose(make(chan *amqp.Error, 1))
-	s.log.Printf("consuming %s, bound to %s/{%s,%s,%s}", ControlQueue, Exchange, KeyEnrol, KeyReport, KeyAlive)
+	s.log.Printf("consuming %s, bound to %s/{%s,%s,%s,%s}",
+		ControlQueue, Exchange, KeyEnrol, KeyReport, KeyAlive, KeyBuilt)
 
 	for {
 		select {
@@ -149,6 +168,8 @@ func (s *Server) handle(ctx context.Context, delivery amqp.Delivery) {
 		s.handleReport(delivery)
 	case KeyAlive:
 		s.handleAlive(delivery)
+	case KeyBuilt:
+		s.handleBuilt(ctx, delivery)
 	default:
 		// Rejected without requeue: a message nothing understands will not be understood on the
 		// next attempt either, and requeuing it would spin.
@@ -255,3 +276,36 @@ func (s *Server) reply(ctx context.Context, delivery amqp.Delivery, reply EnrolR
 		s.log.Printf("cannot reply to %s: %v", delivery.ReplyTo, err)
 	}
 }
+
+// handleBuilt keeps what a builder said, whichever way it went.
+//
+// This is for results nobody was waiting for. A build asked for with `build` is answered directly
+// to the asker; one triggered any other way is published here, and without this it would be
+// reported into the void — which is the same as not reporting it.
+func (s *Server) handleBuilt(ctx context.Context, delivery amqp.Delivery) {
+	var result BuildResult
+	if err := json.Unmarshal(delivery.Body, &result); err != nil {
+		s.log.Printf("a build result could not be read: %v", err)
+		_ = delivery.Reject(false)
+		return
+	}
+	if s.recorder == nil {
+		// Nothing to keep it in. Rejected rather than dropped silently, so the broker's own
+		// counters show something arriving that nothing handles.
+		s.log.Printf("a build result arrived and this control plane keeps none")
+		_ = delivery.Reject(false)
+		return
+	}
+	if err := s.recorder.Built(ctx, result); err != nil {
+		s.log.Printf("cannot keep a build result from %s: %v", result.On, err)
+		_ = delivery.Reject(false)
+		return
+	}
+	switch {
+	case result.Failed != "":
+		s.log.Printf("%s could not build %s", result.On, result.Repository)
+	default:
+		s.log.Printf("%s built %s from %s", result.On, result.Repository, result.Commit)
+	}
+	_ = delivery.Ack(false)
+}