fix: Fixes to running on kubernetes

server/internal/services/audit_retention.go

@@ -15,13 +15,21 @@ import (
 // An immediate pass then daily, following StartLogSweeper's shape. Daily rather
 // than hourly because the unit of retention is a day: sweeping twenty-four times
 // to delete the same nothing is load without a purpose.
-func StartAuditSweeper() {
+// It takes a context because it runs under leader election: several replicas
+// all trimming the same audit logs is duplicated deletion of customer data, and
+// the loop must stop the moment this process stops being the leader.
+func StartAuditSweeper(ctx context.Context) {
 	go func() {
 		sweepAuditLogs()
 		t := time.NewTicker(24 * time.Hour)
 		defer t.Stop()
-		for range t.C {
-			sweepAuditLogs()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-t.C:
+				sweepAuditLogs()
+			}
 		}
 	}()
 }

server/internal/services/consoleproxy.go

@@ -1,12 +1,17 @@
 package services
 
 import (
+	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"log"
 	"net"
 	"os"
+	"sync"
+	"time"
 
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/bus"
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/grpc/pb"
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/proxy"
 )
@@ -16,23 +21,51 @@ import (
 // so this is fatal rather than a degraded mode.
 var ErrAgentOffline = errors.New("agent is not connected")
 
-// ConsoleProxy is a pending relay: a bound listener guacd can dial and a
-// dispatched command telling the agent to meet it.
+// A console session spans two processes once there is more than one replica.
+//
+// The browser's WebSocket lands on an arbitrary pod. The agent's ProxyStream
+// lands on the pod holding that agent's command stream. The relay listener has
+// to be on the latter — that is the only process that can match an incoming
+// ProxyStream to a waiting listener — while guacd is dialled from the former.
+//
+// So the WebSocket's pod asks, over the bus, for a relay to be bound on the
+// agent's pod, and gets back an address to hand to guacd. That address is the
+// owner pod's own, which is why it must resolve to a single pod (POD_IP under
+// Kubernetes) rather than to the Service, which would send guacd to a pod
+// holding no listener roughly (n-1)/n of the time.
+//
+// Teardown needs no message of its own. When the browser goes away guac closes
+// its connection to the relay, the relay sees the read end, and the session
+// closes itself — the same path a single-process deployment always took. Only
+// the *reason* has to cross back, because the pod that writes the audit event
+// is not the pod that observed the failure.
+
+// How long Close waits for the relay's terminal reason to arrive before giving
+// up. The audit event is written immediately afterwards; a reason that has not
+// crossed the bus in this long is not going to improve the record by being
+// waited for longer.
+const proxyEndGrace = 2 * time.Second
+
+// ConsoleProxy is a relay as seen by the pod serving the WebSocket.
 type ConsoleProxy struct {
 	ProxyID string
 	Host    string
 	Port    int
 
-	session *proxy.Session
+	serverID string
+
+	mu     sync.Mutex
+	reason string
+	closed bool
+	stop   func()
+	ended  chan struct{}
 }
 
-func (c *ConsoleProxy) Close() {
-	proxy.Default.Remove(c.ProxyID)
-	c.session.Close("")
+// proxyEnd is the terminal event a relay's owner pod publishes.
+type proxyEnd struct {
+	Reason string `json:"reason"`
 }
 
-func (c *ConsoleProxy) Reason() string { return c.session.Reason() }
-
 func proxyListenHost() string {
 	if v := os.Getenv("PROXY_LISTEN_HOST"); v != "" {
 		return v
@@ -40,7 +73,14 @@ func proxyListenHost() string {
 	return "0.0.0.0"
 }
 
+// proxyAdvertiseHost is the address guacd will dial to reach a relay bound by
+// *this* process. POD_IP wins over the configured value because with several
+// replicas the configured value names the Service, and a Service cannot address
+// one pod. The chart sets POD_IP from the downward API.
 func proxyAdvertiseHost() string {
+	if v := os.Getenv("POD_IP"); v != "" {
+		return v
+	}
 	if v := os.Getenv("PROXY_ADVERTISE_HOST"); v != "" {
 		return v
 	}
@@ -74,17 +114,55 @@ func guacdHosts(addr string) []string {
 	return ips
 }
 
-// DispatchOpenProxy tells the agent to dial its own loopback on port and relay
-// it back under proxyID.
-func DispatchOpenProxy(serverID, proxyID string, port uint32) error {
-	return Dispatcher.dispatch(serverID, &pb.ServerCommand{
-		CommandId: proxyID,
-		OpenProxy: &pb.OpenProxyCmd{ProxyId: proxyID, Port: port},
-	})
+// localRelay is a listener bound by this process on behalf of a remote request.
+type localRelay struct {
+	proxyID string
+	host    string
+	port    int
+	session *proxy.Session
 }
 
-// OpenConsoleProxy binds a relay listener, registers it, and asks the agent to
-// connect. The caller must Close the result.
+// openLocalRelay binds a listener here and registers it, so the agent's
+// ProxyStream — which will arrive at this process — can be matched to it.
+// Called on the owner pod, from the dispatch handler.
+func openLocalRelay(instanceID, serverID, proxyID string) (*localRelay, error) {
+	sess, err := proxy.NewSession(proxyListenHost(), guacdHosts(guacdAddr()))
+	if err != nil {
+		return nil, err
+	}
+
+	sess.OnEnd(func(reason string) {
+		ctx, cancel := context.WithTimeout(context.Background(), dispatchAckTimeout)
+		defer cancel()
+		if _, err := bus.Publish(ctx, bus.ProxyEndChannel+proxyID, proxyEnd{Reason: reason}); err != nil {
+			log.Printf("proxy: publish end for %s: %v", proxyID, err)
+		}
+	})
+
+	proxy.Default.Add(&proxy.Entry{
+		ProxyID:    proxyID,
+		InstanceID: instanceID,
+		ServerID:   serverID,
+		Session:    sess,
+	})
+
+	return &localRelay{
+		proxyID: proxyID,
+		host:    proxyAdvertiseHost(),
+		port:    sess.Port(),
+		session: sess,
+	}, nil
+}
+
+// abandon tears down a relay that was bound but whose command never reached the
+// agent, so the listener does not sit out its rendezvous timeout for nothing.
+func (r *localRelay) abandon() {
+	proxy.Default.Remove(r.proxyID)
+	r.session.Close("dispatch_failed")
+}
+
+// OpenConsoleProxy asks the pod holding serverID's stream to bind a relay and
+// tell the agent to meet it. The caller must Close the result.
 func OpenConsoleProxy(instanceID, serverID string, targetPort int) (*ConsoleProxy, error) {
 	if !Dispatcher.IsConnected(serverID) {
 		return nil, ErrAgentOffline
@@ -95,28 +173,85 @@ func OpenConsoleProxy(instanceID, serverID string, targetPort int) (*ConsoleProx
 		return nil, fmt.Errorf("generate proxy id: %w", err)
 	}
 
-	sess, err := proxy.NewSession(proxyListenHost(), guacdHosts(guacdAddr()))
+	// Subscribed before the relay is asked for: a relay that fails immediately
+	// (the agent never claims it, the dial is refused) publishes its reason at
+	// once, and that reason is the whole content of the audit event.
+	cp := &ConsoleProxy{ProxyID: proxyID, serverID: serverID, ended: make(chan struct{})}
+	endCtx, endCancel := context.WithCancel(context.Background())
+	ends, unsub, err := bus.Subscribe(endCtx, bus.ProxyEndChannel+proxyID)
 	if err != nil {
-		return nil, err
+		endCancel()
+		return nil, fmt.Errorf("subscribe relay end: %w", err)
 	}
+	cp.stop = func() {
+		endCancel()
+		unsub()
+	}
+	go cp.watchEnd(ends)
 
-	proxy.Default.Add(&proxy.Entry{
-		ProxyID:    proxyID,
-		InstanceID: instanceID,
-		ServerID:   serverID,
-		Session:    sess,
+	ack, err := Dispatcher.send(CommandEnvelope{
+		ServerID: serverID,
+		Command: &pb.ServerCommand{
+			CommandId: proxyID,
+			OpenProxy: &pb.OpenProxyCmd{ProxyId: proxyID, Port: uint32(targetPort)},
+		},
+		Proxy: &ProxyRelayRequest{InstanceID: instanceID, ProxyID: proxyID},
 	})
-
-	if err := DispatchOpenProxy(serverID, proxyID, uint32(targetPort)); err != nil {
-		proxy.Default.Remove(proxyID)
-		sess.Close("dispatch_failed")
+	if err != nil {
+		cp.stop()
+		if errors.Is(err, ErrAgentNotConnected) {
+			return nil, ErrAgentOffline
+		}
 		return nil, err
 	}
+	if ack.ProxyHost == "" || ack.ProxyPort == 0 {
+		cp.stop()
+		return nil, fmt.Errorf("relay opened without an address")
+	}
 
-	return &ConsoleProxy{
-		ProxyID: proxyID,
-		Host:    proxyAdvertiseHost(),
-		Port:    sess.Port(),
-		session: sess,
-	}, nil
+	cp.Host = ack.ProxyHost
+	cp.Port = ack.ProxyPort
+	return cp, nil
+}
+
+func (c *ConsoleProxy) watchEnd(ends <-chan []byte) {
+	defer close(c.ended)
+	b, ok := <-ends
+	if !ok {
+		return
+	}
+	var e proxyEnd
+	if err := json.Unmarshal(b, &e); err != nil {
+		return
+	}
+	c.mu.Lock()
+	if c.reason == "" {
+		c.reason = e.Reason
+	}
+	c.mu.Unlock()
+}
+
+// Close waits briefly for the relay's terminal reason, then releases the
+// subscription. It is safe to call more than once.
+func (c *ConsoleProxy) Close() {
+	c.mu.Lock()
+	if c.closed {
+		c.mu.Unlock()
+		return
+	}
+	c.closed = true
+	c.mu.Unlock()
+
+	select {
+	case <-c.ended:
+	case <-time.After(proxyEndGrace):
+	}
+	c.stop()
+}
+
+// Reason reports why the relay ended, empty if it ended cleanly.
+func (c *ConsoleProxy) Reason() string {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.reason
 }

server/internal/services/dispatch.go

@@ -1,63 +1,253 @@
 package services
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
+	"log"
 	"net/http"
 	"strings"
-	"sync"
+	"time"
 
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/bus"
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/grpc/pb"
 	"github.com/google/uuid"
 )
 
-type commandDispatcher struct {
-	mu       sync.RWMutex
-	channels map[string]chan *pb.ServerCommand
+// Commands travel over the bus even when the sender is also the owner.
+//
+// An agent's CommandStream terminates on exactly one process, and with several
+// replicas that is almost never the process handling the REST request that
+// wants to talk to it. Publishing unconditionally — rather than checking for a
+// local stream first and falling back — means one code path, exercised on every
+// deployment including the single-replica ones, instead of a rare cross-pod
+// path that only fails in production.
+const (
+	// How long a caller waits for the owning pod to acknowledge. Generous
+	// enough to cross a loaded cluster, short enough that a REST handler
+	// answering 503 does not look hung.
+	dispatchAckTimeout = 5 * time.Second
+
+	// Presence must outlive a renew or two, or a momentarily slow pod would
+	// look offline and its agent would be declared unreachable.
+	presenceTTL   = 30 * time.Second
+	presenceRenew = 10 * time.Second
+)
+
+// CommandEnvelope is what actually crosses the bus. It is the command plus the
+// small amount of context the owning pod needs to act on it locally.
+type CommandEnvelope struct {
+	ServerID string             `json:"server_id"`
+	Command  *pb.ServerCommand  `json:"command"`
+	ReplyTo  string             `json:"reply_to"`
+	Log      *LogRequest        `json:"log,omitempty"`
+	Proxy    *ProxyRelayRequest `json:"proxy,omitempty"`
 }
 
-var Dispatcher = &commandDispatcher{
-	channels: make(map[string]chan *pb.ServerCommand),
+// LogRequest asks the owner pod to open a step log before it dispatches.
+//
+// Step output arrives on the owner pod's gRPC stream, so that is where it is
+// masked and written. Shipping raw output back to the run's pod first would put
+// unmasked bytes on the bus for no gain.
+type LogRequest struct {
+	RunID    string   `json:"run_id"`
+	ServerID string   `json:"server_id"`
+	Mask     []string `json:"mask,omitempty"`
 }
 
-func (d *commandDispatcher) Connect(serverID string) chan *pb.ServerCommand {
-	ch := make(chan *pb.ServerCommand, 16)
-	d.mu.Lock()
-	d.channels[serverID] = ch
-	d.mu.Unlock()
-	return ch
+// ProxyRelayRequest asks the owner pod to bind a console relay listener and
+// register it before dispatching OpenProxyCmd.
+//
+// The listener has to live on the owner pod: the agent's ProxyStream arrives
+// there, and only there can it be matched to a waiting listener. The pod
+// serving the browser's WebSocket learns the address from the ack and hands
+// that to guacd.
+type ProxyRelayRequest struct {
+	InstanceID string `json:"instance_id"`
+	ProxyID    string `json:"proxy_id"`
 }
 
-func (d *commandDispatcher) Disconnect(serverID string) {
-	d.mu.Lock()
-	delete(d.channels, serverID)
-	d.mu.Unlock()
+// CommandAck is the owner pod's answer.
+type CommandAck struct {
+	OK        bool   `json:"ok"`
+	Error     string `json:"error,omitempty"`
+	Node      string `json:"node,omitempty"`
+	ProxyHost string `json:"proxy_host,omitempty"`
+	ProxyPort int    `json:"proxy_port,omitempty"`
 }
 
+type commandDispatcher struct{}
+
+var Dispatcher = &commandDispatcher{}
+
+// ErrAgentNotConnected is returned when no pod holds the agent's stream.
+var ErrAgentNotConnected = errors.New("agent is not connected")
+
+// Serve subscribes this pod to serverID's command channel and claims presence
+// for it, returning the channel the gRPC handler should send from and a
+// teardown function.
+//
+// send is the local delivery: it is called on the bus goroutine and must not
+// block for long, which is why it pushes onto a buffered channel rather than
+// writing to the gRPC stream directly.
+func (d *commandDispatcher) Serve(ctx context.Context, serverID string) (<-chan *pb.ServerCommand, func()) {
+	out := make(chan *pb.ServerCommand, 16)
+
+	envelopes, unsub, err := bus.Subscribe(ctx, bus.CommandChannel+serverID)
+	if err != nil {
+		// Without a subscription this pod cannot receive commands for the
+		// agent. Claiming presence anyway would advertise a stream nobody can
+		// reach, which is worse than the agent appearing offline.
+		log.Printf("dispatch: subscribe for %s failed, commands will not reach it: %v", serverID, err)
+		close(out)
+		return out, func() {}
+	}
+
+	runCtx, cancel := context.WithCancel(ctx)
+
+	if err := bus.SetPresence(runCtx, serverID, presenceTTL); err != nil {
+		log.Printf("dispatch: claim presence for %s: %v", serverID, err)
+	}
+	go func() {
+		t := time.NewTicker(presenceRenew)
+		defer t.Stop()
+		for {
+			select {
+			case <-runCtx.Done():
+				return
+			case <-t.C:
+				if err := bus.SetPresence(runCtx, serverID, presenceTTL); err != nil {
+					log.Printf("dispatch: renew presence for %s: %v", serverID, err)
+				}
+			}
+		}
+	}()
+
+	go func() {
+		for {
+			select {
+			case <-runCtx.Done():
+				return
+			case raw, ok := <-envelopes:
+				if !ok {
+					return
+				}
+				d.handleEnvelope(runCtx, raw, out)
+			}
+		}
+	}()
+
+	return out, func() {
+		cancel()
+		unsub()
+		relCtx, relCancel := context.WithTimeout(context.Background(), 2*time.Second)
+		if err := bus.ClearPresence(relCtx, serverID); err != nil {
+			log.Printf("dispatch: release presence for %s: %v", serverID, err)
+		}
+		relCancel()
+	}
+}
+
+// handleEnvelope performs the owner-pod side of a dispatch: any local setup the
+// command needs, then queueing it for the stream, then the ack.
+func (d *commandDispatcher) handleEnvelope(ctx context.Context, raw []byte, out chan *pb.ServerCommand) {
+	var env CommandEnvelope
+	if err := json.Unmarshal(raw, &env); err != nil {
+		log.Printf("dispatch: undecodable envelope: %v", err)
+		return
+	}
+	if env.Command == nil || env.ReplyTo == "" {
+		return
+	}
+
+	ack := CommandAck{OK: true, Node: bus.NodeID()}
+
+	if env.Log != nil {
+		if err := StepLogs.Open(env.Command.CommandId, env.Log.RunID, env.Log.ServerID, env.Log.Mask); err != nil {
+			log.Printf("dispatch: open step log for %s: %v", env.Command.CommandId, err)
+		}
+	}
+
+	var relay *localRelay
+	if env.Proxy != nil {
+		r, err := openLocalRelay(env.Proxy.InstanceID, env.ServerID, env.Proxy.ProxyID)
+		if err != nil {
+			ack = CommandAck{OK: false, Error: err.Error(), Node: bus.NodeID()}
+		} else {
+			relay = r
+			ack.ProxyHost = r.host
+			ack.ProxyPort = r.port
+		}
+	}
+
+	if ack.OK {
+		select {
+		case out <- env.Command:
+		default:
+			ack = CommandAck{OK: false, Error: "command queue full", Node: bus.NodeID()}
+			if relay != nil {
+				relay.abandon()
+			}
+		}
+	}
+
+	if err := bus.Reply(ctx, env.ReplyTo, ack); err != nil {
+		log.Printf("dispatch: reply on %s: %v", env.ReplyTo, err)
+	}
+}
+
+// IsConnected reports whether any pod holds the agent's command stream.
 func (d *commandDispatcher) IsConnected(serverID string) bool {
-	d.mu.RLock()
-	_, ok := d.channels[serverID]
-	d.mu.RUnlock()
-	return ok
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	return bus.IsConnected(ctx, serverID)
 }
 
 func (d *commandDispatcher) dispatch(serverID string, cmd *pb.ServerCommand) error {
-	d.mu.RLock()
-	ch, ok := d.channels[serverID]
-	d.mu.RUnlock()
-	if !ok {
-		return fmt.Errorf("agent for server %s is not connected", serverID)
-	}
-	select {
-	case ch <- cmd:
-		return nil
-	default:
-		return fmt.Errorf("command queue full for server %s", serverID)
-	}
+	_, err := d.send(CommandEnvelope{ServerID: serverID, Command: cmd})
+	return err
 }
 
-func DispatchRunStep(serverID, commandID string, cmd *pb.RunStepCmd) error {
-	return Dispatcher.dispatch(serverID, &pb.ServerCommand{CommandId: commandID, RunStep: cmd})
+// send publishes the envelope and waits for the owning pod's ack. A missing
+// responder and a refusing responder are both errors: in neither case did the
+// command reach the agent.
+func (d *commandDispatcher) send(env CommandEnvelope) (CommandAck, error) {
+	if env.Command == nil || env.Command.CommandId == "" {
+		return CommandAck{}, fmt.Errorf("command id is required")
+	}
+	env.ReplyTo = bus.AckChannelFor(env.Command.CommandId)
+
+	ctx, cancel := context.WithTimeout(context.Background(), dispatchAckTimeout)
+	defer cancel()
+
+	raw, err := bus.Request(ctx, bus.CommandChannel+env.ServerID, env.ReplyTo, env, dispatchAckTimeout)
+	if err != nil {
+		if errors.Is(err, bus.ErrNoResponder) {
+			return CommandAck{}, fmt.Errorf("%w: %s", ErrAgentNotConnected, env.ServerID)
+		}
+		return CommandAck{}, err
+	}
+
+	var ack CommandAck
+	if err := json.Unmarshal(raw, &ack); err != nil {
+		return CommandAck{}, fmt.Errorf("undecodable ack: %w", err)
+	}
+	if !ack.OK {
+		return ack, fmt.Errorf("agent for server %s: %s", env.ServerID, ack.Error)
+	}
+	return ack, nil
+}
+
+// DispatchRunStep sends a step to the agent and asks the owning pod to capture
+// its output. mask holds the secret values that must not reach the log.
+func DispatchRunStep(serverID, commandID, runID string, mask []string, cmd *pb.RunStepCmd) error {
+	_, err := Dispatcher.send(CommandEnvelope{
+		ServerID: serverID,
+		Command:  &pb.ServerCommand{CommandId: commandID, RunStep: cmd},
+		Log:      &LogRequest{RunID: runID, ServerID: serverID, Mask: mask},
+	})
+	return err
 }
 
 func DispatchCleanupWorkspace(serverID, workspaceID string) {

server/internal/services/steplogs.go

@@ -2,9 +2,8 @@ package services
 
 import (
 	"bytes"
+	"context"
 	"log"
-	"os"
-	"path/filepath"
 	"strings"
 	"sync"
 	"time"
@@ -12,47 +11,130 @@ import (
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/db"
 	"go.mongodb.org/mongo-driver/v2/bson"
 	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
 )
 
-func WorkflowLogDir() string {
-	dir := os.Getenv("VANTAGE_WORKFLOW_LOG_DIR")
-	if dir == "" {
-		dir = filepath.Join("data", "workflow-logs")
-	}
-	_ = os.MkdirAll(dir, 0700)
-	return dir
-}
+// Workflow run logs live in MongoDB, not on disk.
+//
+// They used to be a file per (run, server) under a volume. That is the cheapest
+// possible writer, and it is wrong the moment there is more than one server
+// process: step output arrives on whichever pod holds the agent's stream, the
+// run's markers are written by whichever pod started the run, and the browser
+// asks for the log through whichever pod the load balancer picked. Three pods,
+// one file, one local disk — two of them see an empty log.
+//
+// Mongo makes every pod an equal reader and writer, which is the property that
+// matters. It costs writes on the hot path, so the writer batches (see
+// stepLogWriter) and both caps below exist to keep a runaway step from turning
+// a workflow into a database incident.
+const (
+	// A single line longer than this is truncated. Base64 blobs and minified
+	// output are the usual cause; nobody reads column 9000 of a log line.
+	maxLogLineBytes = 8 * 1024
 
-func ServerRunLogPath(runID, serverID string) string {
-	return filepath.Join(WorkflowLogDir(), runID, serverID+".log")
+	// A single (run, server) log stops accepting lines here, with one final
+	// marker saying so. 200k lines is far past what anyone reads and still
+	// only a few tens of MB. Without a cap, `yes` in a step fills the database.
+	maxLogLines = 200_000
+
+	// The writer flushes on whichever comes first. Batching is what keeps a
+	// chatty step to a handful of writes a second instead of one per line.
+	logFlushLines    = 128
+	logFlushInterval = 250 * time.Millisecond
+)
+
+const logLinesCol = "workflow_log_lines"
+const logSeqCol = "workflow_log_seq"
+
+func logCtx() (context.Context, context.CancelFunc) {
+	return context.WithTimeout(context.Background(), 10*time.Second)
 }
 
 func logTS() string {
 	return time.Now().UTC().Format("2006-01-02T15:04:05.000") + "Z"
 }
 
-func AppendMarker(runID, serverID, text string) (int64, error) {
-	path := ServerRunLogPath(runID, serverID)
-	if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
-		return 0, err
+func seqID(runID, serverID string) string { return runID + "/" + serverID }
+
+// reserveSeq atomically claims n consecutive sequence numbers for a
+// (run, server) log and returns the first.
+//
+// The counter is a document rather than a per-process integer because two pods
+// write the same log concurrently: the run's pod emits markers while the
+// agent's pod emits step output. Ordering between them is only meaningful if
+// they draw from the same counter.
+func reserveSeq(ctx context.Context, runID, serverID string, n int) (int64, error) {
+	var doc struct {
+		Seq int64 `bson:"seq"`
 	}
-	f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600)
+	err := db.Col(logSeqCol).FindOneAndUpdate(ctx,
+		bson.M{"_id": seqID(runID, serverID)},
+		bson.M{"$inc": bson.M{"seq": int64(n)}},
+		options.FindOneAndUpdate().SetUpsert(true).SetReturnDocument(options.After),
+	).Decode(&doc)
 	if err != nil {
 		return 0, err
 	}
-	defer f.Close()
-	off, _ := f.Seek(0, 2)
-	if _, err := f.WriteString("[" + logTS() + "] " + text + "\n"); err != nil {
-		return off, err
-	}
-	return off, nil
+	return doc.Seq - int64(n), nil
 }
 
+type logLine struct {
+	RunID    string    `bson:"run_id"`
+	ServerID string    `bson:"server_id"`
+	Seq      int64     `bson:"seq"`
+	At       time.Time `bson:"at"`
+	Line     string    `bson:"line"`
+}
+
+// writeLines reserves a block of sequence numbers and inserts the batch.
+func writeLines(ctx context.Context, runID, serverID string, lines []string) (int64, error) {
+	if len(lines) == 0 {
+		return 0, nil
+	}
+	start, err := reserveSeq(ctx, runID, serverID, len(lines))
+	if err != nil {
+		return 0, err
+	}
+	now := time.Now().UTC()
+	docs := make([]any, 0, len(lines))
+	for i, l := range lines {
+		docs = append(docs, logLine{
+			RunID:    runID,
+			ServerID: serverID,
+			Seq:      start + int64(i),
+			At:       now,
+			Line:     l,
+		})
+	}
+	// Unordered: one rejected document must not discard the rest of the batch.
+	_, err = db.Col(logLinesCol).InsertMany(ctx, docs, options.InsertMany().SetOrdered(false))
+	return start, err
+}
+
+// AppendMarker writes one control line (step banners, retries, run outcome) and
+// returns its sequence number, which is what a StepRun's log_offset records so
+// the UI can scroll to where a step began.
+func AppendMarker(runID, serverID, text string) (int64, error) {
+	ctx, cancel := logCtx()
+	defer cancel()
+	return writeLines(ctx, runID, serverID, []string{"[" + logTS() + "] " + text})
+}
+
+// stepLogWriter accumulates one command's output, splits it into lines, masks
+// secrets and flushes batches to Mongo.
 type stepLogWriter struct {
-	mu      sync.Mutex
-	f       *os.File
+	mu       sync.Mutex
+	runID    string
+	serverID string
+	secrets  []string
+
 	carry   []byte
-	secrets []string
+	pending []string
+	written int
+	capped  bool
+
+	stop chan struct{}
+	done chan struct{}
 }
 
 type stepLogRegistry struct {
@@ -60,18 +142,30 @@ type stepLogRegistry struct {
 	writers map[string]*stepLogWriter
 }
 
+// The registry stays process-local, and correctly so: a step's output arrives
+// on the pod holding that agent's stream, and that is the same pod the
+// dispatch envelope asked to open the writer. Nothing here crosses pods —
+// only the lines it produces do, by virtue of landing in Mongo.
 var StepLogs = &stepLogRegistry{writers: make(map[string]*stepLogWriter)}
 
-func (r *stepLogRegistry) Open(commandID, path string, secrets []string) error {
-	if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {
-		return err
+func (r *stepLogRegistry) Open(commandID, runID, serverID string, secrets []string) error {
+	w := &stepLogWriter{
+		runID:    runID,
+		serverID: serverID,
+		secrets:  secrets,
+		stop:     make(chan struct{}),
+		done:     make(chan struct{}),
 	}
-	f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600)
-	if err != nil {
-		return err
-	}
-	w := &stepLogWriter{f: f, secrets: secrets}
+	go w.flushLoop()
+
 	r.mu.Lock()
+	if old := r.writers[commandID]; old != nil {
+		// A retry reuses the command ID. Close the previous attempt's writer
+		// rather than leaking its flush goroutine.
+		r.mu.Unlock()
+		old.close()
+		r.mu.Lock()
+	}
 	r.writers[commandID] = w
 	r.mu.Unlock()
 	return nil
@@ -89,24 +183,22 @@ func (r *stepLogRegistry) Append(commandID string, data []byte) {
 		return
 	}
 	w.mu.Lock()
-	defer w.mu.Unlock()
 	buf := append(w.carry, data...)
 	for {
 		i := bytes.IndexByte(buf, '\n')
 		if i < 0 {
 			break
 		}
-		w.writeLine(buf[:i])
+		w.queueLine(buf[:i])
 		buf = buf[i+1:]
 	}
 	w.carry = append([]byte{}, buf...)
-}
+	full := len(w.pending) >= logFlushLines
+	w.mu.Unlock()
 
-func (w *stepLogWriter) writeLine(line []byte) {
-	masked := maskBytes(line, w.secrets)
-	_, _ = w.f.WriteString("[" + logTS() + "] ")
-	_, _ = w.f.Write(masked)
-	_, _ = w.f.WriteString("\n")
+	if full {
+		w.flush()
+	}
 }
 
 func (r *stepLogRegistry) Close(commandID string) {
@@ -117,13 +209,72 @@ func (r *stepLogRegistry) Close(commandID string) {
 	if w == nil {
 		return
 	}
+	w.close()
+}
+
+// queueLine masks, truncates and enqueues a line. Caller holds w.mu.
+func (w *stepLogWriter) queueLine(line []byte) {
+	if w.capped {
+		return
+	}
+	if w.written+len(w.pending) >= maxLogLines {
+		w.capped = true
+		w.pending = append(w.pending, "["+logTS()+"] [vantage] log truncated: this step exceeded the per-server line limit")
+		return
+	}
+	masked := maskBytes(line, w.secrets)
+	if len(masked) > maxLogLineBytes {
+		masked = append(masked[:maxLogLineBytes], []byte(" [truncated]")...)
+	}
+	w.pending = append(w.pending, "["+logTS()+"] "+string(masked))
+}
+
+func (w *stepLogWriter) flush() {
+	w.mu.Lock()
+	if len(w.pending) == 0 {
+		w.mu.Unlock()
+		return
+	}
+	batch := w.pending
+	w.pending = nil
+	w.written += len(batch)
+	runID, serverID := w.runID, w.serverID
+	w.mu.Unlock()
+
+	ctx, cancel := logCtx()
+	defer cancel()
+	if _, err := writeLines(ctx, runID, serverID, batch); err != nil {
+		// Dropped rather than retried. A log line is not worth stalling the
+		// step it describes, and a Mongo that cannot take writes has larger
+		// problems than a missing line.
+		log.Printf("step log: write %d line(s) for run %s: %v", len(batch), runID, err)
+	}
+}
+
+func (w *stepLogWriter) flushLoop() {
+	defer close(w.done)
+	t := time.NewTicker(logFlushInterval)
+	defer t.Stop()
+	for {
+		select {
+		case <-w.stop:
+			return
+		case <-t.C:
+			w.flush()
+		}
+	}
+}
+
+func (w *stepLogWriter) close() {
+	close(w.stop)
+	<-w.done
 	w.mu.Lock()
-	defer w.mu.Unlock()
 	if len(w.carry) > 0 {
-		w.writeLine(w.carry)
+		w.queueLine(w.carry)
 		w.carry = nil
 	}
-	_ = w.f.Close()
+	w.mu.Unlock()
+	w.flush()
 }
 
 func maskBytes(b []byte, secrets []string) []byte {
@@ -137,86 +288,132 @@ func maskBytes(b []byte, secrets []string) []byte {
 	return []byte(s)
 }
 
-func StartLogSweeper() {
+// ReadServerRunLog returns the log for one server in a run, from sequence
+// number after onwards, along with the highest sequence returned.
+//
+// Callers page with it rather than fetching everything: the live stream asks
+// repeatedly for what is new, and the plain-text endpoint walks the whole log
+// in chunks so a very large one is never held in memory whole.
+func ReadServerRunLog(runID, serverID string, after int64, limit int) ([]string, int64, error) {
+	ctx, cancel := logCtx()
+	defer cancel()
+
+	cur, err := db.Col(logLinesCol).Find(ctx,
+		bson.M{"run_id": runID, "server_id": serverID, "seq": bson.M{"$gt": after}},
+		options.Find().SetSort(bson.D{{Key: "seq", Value: 1}}).SetLimit(int64(limit)),
+	)
+	if err != nil {
+		return nil, after, err
+	}
+	defer cur.Close(ctx)
+
+	lines := make([]string, 0, 64)
+	last := after
+	for cur.Next(ctx) {
+		var l logLine
+		if err := cur.Decode(&l); err != nil {
+			return lines, last, err
+		}
+		lines = append(lines, l.Line)
+		last = l.Seq
+	}
+	return lines, last, cur.Err()
+}
+
+// HasServerRunLog reports whether any line exists, so a log read can answer 404
+// rather than 200 with an empty body.
+func HasServerRunLog(runID, serverID string) bool {
+	ctx, cancel := logCtx()
+	defer cancel()
+	err := db.Col(logLinesCol).FindOne(ctx, bson.M{"run_id": runID, "server_id": serverID}).Err()
+	return err == nil
+}
+
+func StartLogSweeper(ctx context.Context) {
 	go func() {
 		sweepLogs()
 		t := time.NewTicker(time.Hour)
 		defer t.Stop()
-		for range t.C {
-			sweepLogs()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-t.C:
+				sweepLogs()
+			}
 		}
 	}()
 }
 
+// sweepLogs deletes the lines of finished runs past their instance's retention.
+//
+// It walks workflow_runs rather than the log collection: retention is a
+// property of the run (its instance, its finish time), and a run row is the
+// only place both are recorded.
 func sweepLogs() {
-	base := WorkflowLogDir()
-	entries, err := os.ReadDir(base)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+	defer cancel()
+
+	cur, err := db.Col("workflow_runs").Find(ctx,
+		bson.M{"finished_at": bson.M{"$ne": nil}},
+		options.Find().SetProjection(bson.M{"run_id": 1, "instance_id": 1, "finished_at": 1}),
+	)
 	if err != nil {
+		log.Printf("log sweep: list runs: %v", err)
 		return
 	}
+	defer cur.Close(ctx)
+
 	cache := map[string]int{}
 	now := time.Now()
 
-	for _, e := range entries {
-		if !e.IsDir() {
-			continue
+	for cur.Next(ctx) {
+		var run struct {
+			RunID      string     `bson:"run_id"`
+			InstanceID string     `bson:"instance_id"`
+			FinishedAt *time.Time `bson:"finished_at"`
 		}
-		runID := e.Name()
-		dir := filepath.Join(base, runID)
-
-		instanceID, finishedAt, found, err := runRetentionInfo(runID)
-		if err != nil {
-
-			log.Printf("log sweep: retention lookup failed for run %s: %v", runID, err)
-			continue
-		}
-		if found && finishedAt == nil {
+		if err := cur.Decode(&run); err != nil || run.FinishedAt == nil {
 			continue
 		}
 
-		days, ok := cache[instanceID]
+		days, ok := cache[run.InstanceID]
 		if !ok {
 			days = defaultRetentionDays
-			if instanceID != "" {
-				if v, err := GetWorkflowLogRetentionDays(instanceID); err == nil {
+			if run.InstanceID != "" {
+				if v, err := GetWorkflowLogRetentionDays(run.InstanceID); err == nil {
 					days = v
 				}
 			}
-			cache[instanceID] = days
+			cache[run.InstanceID] = days
 		}
 		if days <= 0 {
 			continue
 		}
-		cutoff := now.AddDate(0, 0, -days)
-
-		if found {
-			if finishedAt.Before(cutoff) {
-				_ = os.RemoveAll(dir)
-			}
+		if !run.FinishedAt.Before(now.AddDate(0, 0, -days)) {
 			continue
 		}
 
-		if fi, e := os.Stat(dir); e == nil && fi.ModTime().Before(cutoff) {
-			_ = os.RemoveAll(dir)
+		if _, err := db.Col(logLinesCol).DeleteMany(ctx, bson.M{"run_id": run.RunID}); err != nil {
+			log.Printf("log sweep: delete lines for run %s: %v", run.RunID, err)
+			continue
 		}
+		_, _ = db.Col(logSeqCol).DeleteMany(ctx, bson.M{"_id": bson.M{"$regex": "^" + run.RunID + "/"}})
 	}
 }
 
 const defaultRetentionDays = 30
 
-func runRetentionInfo(runID string) (string, *time.Time, bool, error) {
-	ctx, cancel := wfCtx()
+// EnsureLogIndexes builds the indexes the log store depends on. The compound
+// index is not an optimisation: every read is a range scan over it, and without
+// it a log read collection-scans every line in the database.
+func EnsureLogIndexes() error {
+	ctx, cancel := logCtx()
 	defer cancel()
-	var run struct {
-		InstanceID string     `bson:"instance_id"`
-		FinishedAt *time.Time `bson:"finished_at"`
+	if _, err := db.Col(logLinesCol).Indexes().CreateOne(ctx, mongo.IndexModel{
+		Keys: bson.D{{Key: "run_id", Value: 1}, {Key: "server_id", Value: 1}, {Key: "seq", Value: 1}},
+	}); err != nil {
+		return err
 	}
-	err := db.Col("workflow_runs").FindOne(ctx, bson.M{"run_id": runID}).Decode(&run)
-	if err == mongo.ErrNoDocuments {
-		return "", nil, false, nil
-	}
-	if err != nil {
-		return "", nil, false, err
-	}
-	return run.InstanceID, run.FinishedAt, true, nil
+	return nil
 }

server/internal/services/stepresults.go

@@ -1,43 +1,75 @@
 package services
 
 import (
-	"sync"
+	"context"
+	"encoding/json"
+	"log"
 
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/bus"
 	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/grpc/pb"
 )
 
-type stepResultRegistry struct {
-	mu      sync.Mutex
-	pending map[string]chan *pb.StepResult
-}
-
-var StepResults = &stepResultRegistry{pending: make(map[string]chan *pb.StepResult)}
-
-func (r *stepResultRegistry) Await(commandID string) <-chan *pb.StepResult {
-	ch := make(chan *pb.StepResult, 1)
-	r.mu.Lock()
-	r.pending[commandID] = ch
-	r.mu.Unlock()
-	return ch
-}
-
-func (r *stepResultRegistry) Cancel(commandID string) {
-	r.mu.Lock()
-	delete(r.pending, commandID)
-	r.mu.Unlock()
+// Step results travel back over the bus for the same reason commands travel out
+// over it: the pod driving a workflow run and the pod holding the agent's
+// stream are two different processes, and a map in one of them cannot be read
+// by the other.
+//
+// Await subscribes before the command is dispatched (see dispatchAndWait),
+// which is what stops a fast agent from answering into a channel nobody is
+// listening on yet.
+
+type stepResultRegistry struct{}
+
+var StepResults = &stepResultRegistry{}
+
+// Await subscribes to a command's result channel. The returned cancel function
+// must be called once the caller is done, whether a result arrived or not —
+// it is what releases the Redis subscription.
+func (r *stepResultRegistry) Await(commandID string) (<-chan *pb.StepResult, func()) {
+	out := make(chan *pb.StepResult, 1)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	raw, unsub, err := bus.Subscribe(ctx, bus.ResultChannel+commandID)
+	if err != nil {
+		log.Printf("step results: subscribe for %s: %v", commandID, err)
+		cancel()
+		close(out)
+		return out, func() {}
+	}
+
+	go func() {
+		defer close(out)
+		select {
+		case <-ctx.Done():
+			return
+		case b, ok := <-raw:
+			if !ok {
+				return
+			}
+			var res pb.StepResult
+			if err := json.Unmarshal(b, &res); err != nil {
+				log.Printf("step results: undecodable result for %s: %v", commandID, err)
+				return
+			}
+			out <- &res
+		}
+	}()
+
+	return out, func() {
+		cancel()
+		unsub()
+	}
 }
 
+// Deliver publishes a result received from an agent. Called on the pod holding
+// that agent's stream, which is not usually the pod waiting for it.
 func (r *stepResultRegistry) Deliver(res *pb.StepResult) {
-	if res == nil {
+	if res == nil || res.CommandId == "" {
 		return
 	}
-	r.mu.Lock()
-	ch, ok := r.pending[res.CommandId]
-	if ok {
-		delete(r.pending, res.CommandId)
-	}
-	r.mu.Unlock()
-	if ok {
-		ch <- res
+	ctx, cancel := context.WithTimeout(context.Background(), dispatchAckTimeout)
+	defer cancel()
+	if _, err := bus.Publish(ctx, bus.ResultChannel+res.CommandId, res); err != nil {
+		log.Printf("step results: publish for %s: %v", res.CommandId, err)
 	}
 }

server/internal/services/workflow_runner.go

@@ -240,7 +240,6 @@ func runServer(instanceID, runID string, srvIdx int, steps []models.ResolvedStep
 
 		marker := fmt.Sprintf("===== step %d/%d: %s (%s) =====", step.Order+1, len(steps), step.Name, step.Interpreter)
 		offset, _ := AppendMarker(runID, serverID, marker)
-		logPath := ServerRunLogPath(runID, serverID)
 		secretsSlice := secretValues(secretVals)
 
 		commandID := uuid.New().String()
@@ -250,15 +249,17 @@ func runServer(instanceID, runID string, srvIdx int, steps []models.ResolvedStep
 				_, _ = AppendMarker(runID, serverID, fmt.Sprintf("retry %d/%d after failure", attempts-1, maxAttempts-1))
 			}
 
-			_ = StepLogs.Open(commandID, logPath, secretsSlice)
-			res = dispatchAndWait(serverID, commandID, &pb.RunStepCmd{
+			// The log writer is opened by the pod that owns this agent's
+			// stream, not here: that is where the output arrives, and the mask
+			// list travels with the dispatch so it is applied before anything
+			// is stored.
+			res = dispatchAndWait(serverID, commandID, runID, secretsSlice, &pb.RunStepCmd{
 				Interpreter:    step.Interpreter,
 				Script:         step.Script,
 				Env:            cmdEnv,
 				TimeoutSeconds: 0,
 				WorkspaceId:    runID,
 			})
-			StepLogs.Close(commandID)
 			if res != nil && res.ExitCode == 0 {
 				break
 			}
@@ -322,10 +323,14 @@ func runServer(instanceID, runID string, srvIdx int, steps []models.ResolvedStep
 	})
 }
 
-func dispatchAndWait(serverID, commandID string, cmd *pb.RunStepCmd) *pb.StepResult {
-	ch := StepResults.Await(commandID)
-	if err := DispatchRunStep(serverID, commandID, cmd); err != nil {
-		StepResults.Cancel(commandID)
+// dispatchAndWait subscribes to the result before dispatching, because the two
+// happen on different pods and an agent that answers quickly would otherwise
+// publish into a channel this process had not yet joined.
+func dispatchAndWait(serverID, commandID, runID string, mask []string, cmd *pb.RunStepCmd) *pb.StepResult {
+	ch, done := StepResults.Await(commandID)
+	defer done()
+
+	if err := DispatchRunStep(serverID, commandID, runID, mask, cmd); err != nil {
 		return &pb.StepResult{ExitCode: 1, Stderr: "[vantage] dispatch failed: " + err.Error()}
 	}
 	wait := time.Duration(cmd.TimeoutSeconds)*time.Second + stepDispatchGrace
@@ -333,10 +338,12 @@ func dispatchAndWait(serverID, commandID string, cmd *pb.RunStepCmd) *pb.StepRes
 		wait = 30*time.Minute + stepDispatchGrace
 	}
 	select {
-	case res := <-ch:
+	case res, ok := <-ch:
+		if !ok || res == nil {
+			return &pb.StepResult{ExitCode: 1, Stderr: "[vantage] lost the result channel before the agent answered"}
+		}
 		return res
 	case <-time.After(wait):
-		StepResults.Cancel(commandID)
 		return &pb.StepResult{ExitCode: 124, Stderr: "[vantage] timed out waiting for agent result"}
 	}
 }

server/internal/services/workflows.go

@@ -60,6 +60,9 @@ func EnsureWorkflowIndexes() error {
 	}); err != nil {
 		return err
 	}
+	if err := EnsureLogIndexes(); err != nil {
+		return err
+	}
 	_, err := db.Col("workflow_runs").Indexes().CreateOne(ctx, mongo.IndexModel{
 		Keys: bson.D{{Key: "run_id", Value: 1}}, Options: options.Index().SetUnique(true),
 	})