From 8699dc5b7e0e05148d546d37147631e4637c8fdb Mon Sep 17 00:00:00 2001 From: mrhid6 Date: Fri, 31 Jul 2026 16:58:50 +0100 Subject: [PATCH] fix: Renew presence on sub/pub --- claude.md | 2 +- server/internal/bus/bus.go | 22 +++++++++++++++++ server/internal/services/dispatch.go | 36 ++++++++++++++++++++++++++-- 3 files changed, 57 insertions(+), 3 deletions(-) diff --git a/claude.md b/claude.md index 2e1f035..9a1680a 100644 --- a/claude.md +++ b/claude.md @@ -178,7 +178,7 @@ rare cross-pod branch that only fails under load. | Concern | How it crosses replicas | | ---------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | Which pod owns an agent | `vantage:agent:` holds the owner's node ID with a 30s TTL, renewed every 10s. `Dispatcher.IsConnected` is an `EXISTS` on it | -| Sending a command | published to `vantage:cmd:`; the owner pod acks on `vantage:ack:`. **Request/ack, not a queue** — a command whose owner died must fail loudly (503) rather than queue | +| Sending a command | published to `vantage:cmd:`; the owner pod acks on `vantage:ack:`. **Request/ack, not a queue** — a command whose owner died must fail loudly (503) rather than queue. The envelope carries `node`, the presence holder resolved at publish time, and a pod ignores envelopes addressed elsewhere: the channel is a fan-out, and during a reconnect a half-open stream's pod is still subscribed. Unaddressed, it could ack first and queue the command onto a dead stream — the operator told it worked, the agent never seeing it. Presence renewal is owner-only (`RenewPresence`) for the same reason: a blind `SET` let the stale pod steal the key back every 10s | | Step results | the owner pod publishes to `vantage:res:`; the pod driving the run subscribes **before** dispatching, or a fast agent answers into a channel nobody has joined | | Step output | never crosses. The dispatch envelope carries the secret mask list, so the owner pod masks and writes lines itself — unmasked bytes stay off the bus | | Console relay | **not routed to the owner pod at all.** A `ProxyStream` is its own HTTP/2 request and an L7 proxy balances requests, not connections, so it does not follow the command stream — the listener therefore cannot be bound in advance. Whichever pod receives the stream binds it and announces **its own** address on `vantage:proxyaddr:`; `vantage:proxypending:` (30s, consumed atomically) is what authorises the claim, and the failure reason comes back on `vantage:proxyend:` | diff --git a/server/internal/bus/bus.go b/server/internal/bus/bus.go index adc7503..40be232 100644 --- a/server/internal/bus/bus.go +++ b/server/internal/bus/bus.go @@ -204,6 +204,21 @@ func SetPresence(ctx context.Context, serverID string, ttl time.Duration) error return rdb.Set(ctx, PresenceKey+serverID, nodeID, ttl).Err() } +// RenewPresence extends serverID's claim, but only while this node still holds +// it, and reports whether it did. +// +// A blind SET here is wrong, not merely untidy. When an agent reconnects, its +// previous stream can stay half-open on another pod for the length of a +// keepalive cycle, and that pod goes on renewing. Two processes then overwrite +// each other's claim every renewal interval and the key names whichever wrote +// last rather than whichever holds the live stream. A superseded pod must lose +// quietly instead. +func RenewPresence(ctx context.Context, serverID string, ttl time.Duration) bool { + n, err := renewPresenceIfOwner.Run(ctx, rdb, + []string{PresenceKey + serverID}, nodeID, int64(ttl/time.Millisecond)).Int64() + return err == nil && n == 1 +} + // ClearPresence releases serverID, but only if this node still holds it. A // blind DEL would let a pod whose stream had already been re-established // elsewhere delete the new owner's claim on its way out. @@ -274,6 +289,13 @@ end return "" `) +var renewPresenceIfOwner = redis.NewScript(` +if redis.call("GET", KEYS[1]) == ARGV[1] then + return redis.call("PEXPIRE", KEYS[1], ARGV[2]) +end +return 0 +`) + var releaseIfOwner = redis.NewScript(` if redis.call("GET", KEYS[1]) == ARGV[1] then return redis.call("DEL", KEYS[1]) diff --git a/server/internal/services/dispatch.go b/server/internal/services/dispatch.go index 14b4047..44d180f 100644 --- a/server/internal/services/dispatch.go +++ b/server/internal/services/dispatch.go @@ -42,6 +42,15 @@ type CommandEnvelope struct { Command *pb.ServerCommand `json:"command"` ReplyTo string `json:"reply_to"` Log *LogRequest `json:"log,omitempty"` + + // Node names the pod this envelope is for: the presence holder at the time + // it was published. The command channel is a fan-out, so during a reconnect + // two pods can be subscribed for one agent — the pod with the live stream, + // and a pod whose stream is half-open and has not yet noticed. Both would + // receive the envelope, and the first to ack wins the request. If that is + // the stale one, the command is queued onto a dead stream and acked OK: the + // operator is told it worked and the agent never sees it. + Node string `json:"node,omitempty"` } // LogRequest asks the owner pod to open a step log before it dispatches. @@ -108,8 +117,13 @@ func (d *commandDispatcher) Serve(ctx context.Context, serverID string) (<-chan 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) + // Renew only while this pod still holds the claim. Losing it + // means a newer stream for the same agent was established + // elsewhere, and this one is a half-open leftover: it must stop + // renewing rather than overwrite the live owner every 10s. + if !bus.RenewPresence(runCtx, serverID, presenceTTL) { + log.Printf("dispatch: presence for %s is held elsewhere, stopping renewal", serverID) + return } } } @@ -152,6 +166,15 @@ func (d *commandDispatcher) handleEnvelope(ctx context.Context, raw []byte, out return } + // Not addressed to this pod: stay silent rather than ack. Answering would + // win the race against the pod that actually holds the agent's stream, and + // the caller would be told a command succeeded that was queued onto a + // stream nobody is reading. Silence lets the real owner answer, or lets the + // request time out as ErrNoResponder, which fails loudly and correctly. + if env.Node != "" && env.Node != bus.NodeID() { + return + } + ack := CommandAck{OK: true, Node: bus.NodeID()} if env.Log != nil { @@ -195,6 +218,15 @@ func (d *commandDispatcher) send(env CommandEnvelope) (CommandAck, error) { ctx, cancel := context.WithTimeout(context.Background(), dispatchAckTimeout) defer cancel() + // Resolved once, here, and carried in the envelope. Reading it at publish + // time rather than letting subscribers self-select is what makes a stale + // subscriber harmless: it will see an envelope addressed elsewhere and + // ignore it. + env.Node = bus.PresenceHolder(ctx, env.ServerID) + if env.Node == "" { + return CommandAck{}, fmt.Errorf("%w: %s", ErrAgentNotConnected, env.ServerID) + } + raw, err := bus.Request(ctx, bus.CommandChannel+env.ServerID, env.ReplyTo, env, dispatchAckTimeout) if err != nil { if errors.Is(err, bus.ErrNoResponder) {