diff --git a/adminsite/lib/api.ts b/adminsite/lib/api.ts index dff5f18..7741805 100644 --- a/adminsite/lib/api.ts +++ b/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; diff --git a/server/internal/bus/bus.go b/server/internal/bus/bus.go index 40be232..9350b9e 100644 --- a/server/internal/bus/bus.go +++ b/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 `) diff --git a/server/internal/grpc/server.go b/server/internal/grpc/server.go index af60999..efc5899 100644 --- a/server/internal/grpc/server.go +++ b/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 diff --git a/server/internal/services/dispatch.go b/server/internal/services/dispatch.go index 44d180f..f7a07f2 100644 --- a/server/internal/services/dispatch.go +++ b/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) {