fix: Fixes to command stream

adminsite/lib/api.ts

@@ -247,13 +247,13 @@ export interface Entitlement {
     updated_at: string;
 }
 
-export interface CustomerUser {
-    user_id: string;
-    account_id: string;
-    email: string;
-    verified_at?: string | null;
-    created_at: string;
-}
+/*
+ * Staff and customer screens read the SAME customer_users row, so they share one
+ * type. There used to be a second, narrower CustomerUser for the staff side; it
+ * silently stopped matching the moment account_role was added to the model, and
+ * a subset type cannot warn about a field it never claimed to have.
+ */
+export type CustomerUser = AccountUser;
 
 export interface AuditEntry {
     actor: string;

server/internal/bus/bus.go

@@ -204,8 +204,31 @@ 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.
+// RenewResult is the outcome of one presence renewal.
+//
+// Four outcomes, not two, because the three ways a renewal can fail to extend
+// an existing claim call for three different responses. Collapsing them into a
+// single false is what made a momentary Redis blip permanent: the renewal loop
+// treated "the key is gone" and "Redis did not answer" as "another pod owns
+// this agent now" and stopped renewing for the life of a stream that was still
+// perfectly healthy, leaving the agent connected and undispatchable.
+type RenewResult int
+
+const (
+	// RenewFailed means Redis could not be reached. Nothing is known about who
+	// holds the claim, so the only safe response is to try again.
+	RenewFailed RenewResult = iota
+	// RenewedOwner means the claim was ours and its TTL was extended.
+	RenewedOwner
+	// RenewedClaim means nobody held the claim and this node took it. Normal
+	// after a Redis restart, failover, eviction or an outage outliving the TTL.
+	RenewedClaim
+	// RenewLost means another node holds the claim. This stream is superseded.
+	RenewLost
+)
+
+// RenewPresence extends serverID's claim while this node holds it, and reclaims
+// it if nobody does.
 //
 // 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
@@ -213,10 +236,25 @@ func SetPresence(ctx context.Context, serverID string, ttl time.Duration) error
 // 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 {
+//
+// Reclaiming an *absent* key is not that case and is safe: absence means no pod
+// is currently advertising the agent, and the caller demonstrably holds a live
+// stream to it. Refusing to reclaim is what leaves an agent unreachable until it
+// happens to reconnect.
+func RenewPresence(ctx context.Context, serverID string, ttl time.Duration) RenewResult {
 	n, err := renewPresenceIfOwner.Run(ctx, rdb,
 		[]string{PresenceKey + serverID}, nodeID, int64(ttl/time.Millisecond)).Int64()
-	return err == nil && n == 1
+	if err != nil {
+		return RenewFailed
+	}
+	switch n {
+	case 1:
+		return RenewedOwner
+	case 2:
+		return RenewedClaim
+	default:
+		return RenewLost
+	}
 }
 
 // ClearPresence releases serverID, but only if this node still holds it. A
@@ -289,9 +327,16 @@ end
 return ""
 `)
 
+// 1 = renewed, 2 = reclaimed an unheld key, 0 = held by another node.
 var renewPresenceIfOwner = redis.NewScript(`
-if redis.call("GET", KEYS[1]) == ARGV[1] then
-	return redis.call("PEXPIRE", KEYS[1], ARGV[2])
+local v = redis.call("GET", KEYS[1])
+if v == ARGV[1] then
+	redis.call("PEXPIRE", KEYS[1], ARGV[2])
+	return 1
+end
+if not v then
+	redis.call("SET", KEYS[1], ARGV[1], "PX", ARGV[2])
+	return 2
 end
 return 0
 `)

server/internal/grpc/server.go

@@ -382,6 +382,17 @@ func (s *vantageServer) CommandStream(stream pb.Vantage_CommandStreamServer) err
 			if beats == 1 {
 				log.Printf("agent %s command stream beating every %s", srv.ServerID, pingInterval)
 			}
+			// A beat the stream accepted is the strongest liveness evidence this
+			// pod has, so it is also used to renew presence. The renewal ticker
+			// proves only that a goroutine still runs; this proves the stream
+			// itself still carries bytes to the agent. A false answer means
+			// another pod has taken the agent over and this stream is a
+			// half-open leftover, which is worth ending here rather than beating
+			// into for as long as the connection survives.
+			if !services.TouchPresence(ctx, srv.ServerID) {
+				log.Printf("agent %s command stream superseded, closing", srv.ServerID)
+				return nil
+			}
 		case cmd, ok := <-ch:
 			if !ok {
 				return nil

server/internal/services/dispatch.go

@@ -30,9 +30,12 @@ const (
 	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.
+	// look offline and its agent would be declared unreachable. Three renewals
+	// inside one TTL, not two: a renewal is also how a claim is recovered after
+	// Redis loses it, and recovery time is what the operator experiences as the
+	// agent being undispatchable.
 	presenceTTL   = 30 * time.Second
-	presenceRenew = 10 * time.Second
+	presenceRenew = 8 * time.Second
 )
 
 // CommandEnvelope is what actually crosses the bus. It is the command plus the
@@ -106,8 +109,11 @@ func (d *commandDispatcher) Serve(ctx context.Context, serverID string) (<-chan
 
 	runCtx, cancel := context.WithCancel(ctx)
 
+	// A failure here is not fatal and is not special-cased: the renewal loop
+	// reclaims an unheld key, so the first tick repairs it. Anything else would
+	// need a second recovery path for a case the loop already covers.
 	if err := bus.SetPresence(runCtx, serverID, presenceTTL); err != nil {
-		log.Printf("dispatch: claim presence for %s: %v", serverID, err)
+		log.Printf("dispatch: claim presence for %s, renewal will retry: %v", serverID, err)
 	}
 	go func() {
 		t := time.NewTicker(presenceRenew)
@@ -117,12 +123,7 @@ func (d *commandDispatcher) Serve(ctx context.Context, serverID string) (<-chan
 			case <-runCtx.Done():
 				return
 			case <-t.C:
-				// 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)
+				if !renewPresence(runCtx, serverID) {
 					return
 				}
 			}
@@ -154,6 +155,46 @@ func (d *commandDispatcher) Serve(ctx context.Context, serverID string) (<-chan
 	}
 }
 
+// renewPresence extends this pod's claim on serverID and reports whether the
+// claim is still this pod's to hold. False means, and only means, that another
+// pod now owns the agent's stream.
+//
+// A Redis failure returns true. It is tempting to read an error as loss and
+// give up, but nothing is known in that moment about who holds the claim, and
+// the stream this pod is serving is demonstrably alive — the caller is either a
+// ticker on that stream or a beat that just succeeded on it. Standing down on a
+// blip is precisely how an agent ends up connected, beating, and unreachable
+// until it happens to reconnect.
+func renewPresence(ctx context.Context, serverID string) bool {
+	switch bus.RenewPresence(ctx, serverID, presenceTTL) {
+	case bus.RenewLost:
+		// A newer stream for this agent exists elsewhere and this one is a
+		// half-open leftover. Stop, rather than overwrite the live owner every
+		// renewal interval and leave the key naming whichever wrote last.
+		log.Printf("dispatch: presence for %s is held elsewhere, standing down", serverID)
+		return false
+	case bus.RenewedClaim:
+		// Nobody held the key — Redis restarted, failed over, evicted it, or was
+		// unreachable for longer than the TTL. Worth a line: it is the only
+		// evidence that presence was lost and recovered rather than never lost.
+		log.Printf("dispatch: reclaimed presence for %s", serverID)
+	case bus.RenewFailed:
+		log.Printf("dispatch: renew presence for %s failed, will retry", serverID)
+	}
+	return true
+}
+
+// TouchPresence renews serverID's claim off the back of something that proves
+// the stream is alive, and reports whether this pod should keep serving it.
+//
+// The renewal ticker proves only that a goroutine is still scheduled. A beat
+// that the agent's stream accepted proves the stream itself still works, which
+// is the thing presence is supposed to advertise. A false answer means this
+// stream has been superseded and the handler should return.
+func TouchPresence(ctx context.Context, serverID string) bool {
+	return renewPresence(ctx, serverID)
+}
+
 // 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) {