From 7352e02342e48e297cf07bf15a71c1220b00fbff Mon Sep 17 00:00:00 2001 From: talksik Date: Thu, 9 Apr 2026 16:36:46 -0700 Subject: [PATCH 1/6] increases timeout --- go/cmd/pusherservice/main.go | 10 +++++----- go/internal/pusher/server.go | 12 +++++++++--- go/k8s/dev/pusher.yaml | 2 +- go/k8s/prod/pusher.yaml | 2 +- 4 files changed, 16 insertions(+), 10 deletions(-) diff --git a/go/cmd/pusherservice/main.go b/go/cmd/pusherservice/main.go index 816bb9d..5d160bd 100644 --- a/go/cmd/pusherservice/main.go +++ b/go/cmd/pusherservice/main.go @@ -52,15 +52,15 @@ func main() { // Pusher core bridge := pusher.NewRedisBridge(pusherRedis, podID) - authorizer := pusher.NewAuthorizer(networkSvc) - hub := pusher.NewHub(bridge, authorizer) - bridge.SetHub(hub) - server := pusher.NewServer(hub, bridge, authSvc) - // Context for graceful shutdown ctx, cancel := context.WithCancel(context.Background()) defer cancel() + authorizer := pusher.NewAuthorizer(networkSvc) + hub := pusher.NewHub(bridge, authorizer) + bridge.SetHub(hub) + server := pusher.NewServer(ctx, hub, bridge, authSvc) + // Start hub event loop go hub.Run(ctx) diff --git a/go/internal/pusher/server.go b/go/internal/pusher/server.go index 37b997f..1b4dd9d 100644 --- a/go/internal/pusher/server.go +++ b/go/internal/pusher/server.go @@ -15,14 +15,17 @@ import ( type Server struct { pbpusher.UnimplementedPusherServiceServer + ctx context.Context // server-scoped context for graceful shutdown hub *Hub bridge *RedisBridge authSvc auth.AuthService } -// NewServer creates a new pusher server. -func NewServer(hub *Hub, bridge *RedisBridge, authSvc auth.AuthService) *Server { +// NewServer creates a new pusher server. The ctx controls the lifetime of all +// WebSocket connections — when cancelled, all connections are closed gracefully. +func NewServer(ctx context.Context, hub *Hub, bridge *RedisBridge, authSvc auth.AuthService) *Server { return &Server{ + ctx: ctx, hub: hub, bridge: bridge, authSvc: authSvc, @@ -59,7 +62,10 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) { slog.Info("websocket connected", "connId", connID, "humanId", session.HumanId) - ctx, cancel := context.WithCancel(r.Context()) + // Use server context, NOT r.Context(). After WebSocket upgrade, the HTTP + // request context can be cancelled by load balancers or Go's HTTP server, + // and nhooyr/websocket permanently closes the conn on any context error. + ctx, cancel := context.WithCancel(s.ctx) defer cancel() // Auto-subscribe to presence channel so this user appears online diff --git a/go/k8s/dev/pusher.yaml b/go/k8s/dev/pusher.yaml index 8548375..3208b18 100644 --- a/go/k8s/dev/pusher.yaml +++ b/go/k8s/dev/pusher.yaml @@ -88,7 +88,7 @@ metadata: name: pusher-backend-policy spec: default: - timeoutSec: 60 + timeoutSec: 3600 connectionDraining: drainingTimeoutSec: 30 targetRef: diff --git a/go/k8s/prod/pusher.yaml b/go/k8s/prod/pusher.yaml index e4649a5..9e5591e 100644 --- a/go/k8s/prod/pusher.yaml +++ b/go/k8s/prod/pusher.yaml @@ -85,7 +85,7 @@ metadata: name: pusher-backend-policy spec: default: - timeoutSec: 60 + timeoutSec: 3600 connectionDraining: drainingTimeoutSec: 30 targetRef: -- 2.54.0 From 11e1ef355e584cdb42e712daf455e6d1fa3582c4 Mon Sep 17 00:00:00 2001 From: talksik Date: Sun, 12 Apr 2026 09:18:35 -0700 Subject: [PATCH 2/6] refactor: reuse constants for redis db --- go/cmd/orion/main.go | 7 +------ go/cmd/pusherservice/main.go | 9 ++------- go/internal/db/{connect.go => postgres.go} | 0 go/internal/db/redis.go | 8 ++++++++ 4 files changed, 11 insertions(+), 13 deletions(-) rename go/internal/db/{connect.go => postgres.go} (100%) create mode 100644 go/internal/db/redis.go diff --git a/go/cmd/orion/main.go b/go/cmd/orion/main.go index 6cd2183..eb1fc22 100644 --- a/go/cmd/orion/main.go +++ b/go/cmd/orion/main.go @@ -27,13 +27,8 @@ import ( "google.golang.org/grpc/credentials/insecure" ) -const ( - // FIX: Use separate redis instance. We start with higher number because use this same instance in helios. - REDIS_DATABASE_FOR_AUTH int = 4 -) - func redisForAuth() *redis.Client { - return internal.ConnectAndTestRedis(REDIS_DATABASE_FOR_AUTH) + return internal.ConnectAndTestRedis(db.RedisDBAuth) } func main() { diff --git a/go/cmd/pusherservice/main.go b/go/cmd/pusherservice/main.go index 5d160bd..2d0c253 100644 --- a/go/cmd/pusherservice/main.go +++ b/go/cmd/pusherservice/main.go @@ -21,11 +21,6 @@ import ( "google.golang.org/grpc" ) -const ( - redisDBAuth = 4 // shared with orion for session validation - redisDBPusher = 5 // dedicated to pusher state (presence, pub/sub) -) - func main() { port := utils.MustGetEnv("PORT") grpcPort := utils.MustGetEnv("GRPC_PORT") @@ -35,10 +30,10 @@ func main() { defer db.Cleanup() // Redis for auth session validation (same DB as orion) - authRedis := internal.ConnectAndTestRedis(redisDBAuth) + authRedis := internal.ConnectAndTestRedis(db.RedisDBAuth) // Redis for pusher state (presence hashes, pub/sub) - pusherRedis := internal.ConnectAndTestRedis(redisDBPusher) + pusherRedis := internal.ConnectAndTestRedis(db.RedisDBPusher) // Services authSvc := auth.NewAuthService(authRedis, nil) // nil aeroSvc — pusher only calls GetSession diff --git a/go/internal/db/connect.go b/go/internal/db/postgres.go similarity index 100% rename from go/internal/db/connect.go rename to go/internal/db/postgres.go diff --git a/go/internal/db/redis.go b/go/internal/db/redis.go new file mode 100644 index 0000000..d663685 --- /dev/null +++ b/go/internal/db/redis.go @@ -0,0 +1,8 @@ +package db + +// Shared database namespaces used across services +// FIX: Use separate redis instance. We start with higher number because use this same instance in helios. +const ( + RedisDBAuth = 4 // auth sessions + RedisDBPusher = 5 // dedicated to pusher state (presence, pub/sub) +) -- 2.54.0 From 6ff2cb77f8f5ec39fee0563f04c85365c1c74043 Mon Sep 17 00:00:00 2001 From: talksik Date: Sun, 12 Apr 2026 09:18:57 -0700 Subject: [PATCH 3/6] avoid closing connection on pong error --- go/internal/pusher/conn.go | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/go/internal/pusher/conn.go b/go/internal/pusher/conn.go index 89d46c7..3cdf2c0 100644 --- a/go/internal/pusher/conn.go +++ b/go/internal/pusher/conn.go @@ -41,15 +41,14 @@ func (c *Conn) ReadPump(ctx context.Context, hub *Hub) { slog.Info("websocket context cancelled", "connId", c.id, "humanId", c.humanID, "error", ctx.Err()) return } - slog.Info("websocket read error", "connId", c.id, "humanId", c.humanID, "error", err) + slog.Warn("websocket read error", "connId", c.id, "humanId", c.humanID, "error", err) return } // Respond to keep-alive pings if string(data) == "ping" { if err := c.ws.Write(ctx, websocket.MessageText, []byte("pong")); err != nil { - slog.Debug("websocket pong write error", "connId", c.id, "error", err) - return + slog.Warn("websocket pong write error", "connId", c.id, "error", err) } continue } -- 2.54.0 From 3ec6a8d635227dda068de74ea2a2b047b3cdda22 Mon Sep 17 00:00:00 2001 From: talksik Date: Sun, 12 Apr 2026 09:19:46 -0700 Subject: [PATCH 4/6] fix: prevent electron from pausing timer executions Our pings could be paused when our app is backgrounded. --- js/src/main.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/js/src/main.ts b/js/src/main.ts index 007bdf4..51256a5 100644 --- a/js/src/main.ts +++ b/js/src/main.ts @@ -36,6 +36,7 @@ const createWindow = () => { frame: false, webPreferences: { preload: path.join(__dirname, 'preload.js'), + backgroundThrottling: false, }, }); -- 2.54.0 From bb66bc49506019d5607f67ebb3ec2cd2cc26bc4e Mon Sep 17 00:00:00 2001 From: talksik Date: Sun, 12 Apr 2026 09:36:53 -0700 Subject: [PATCH 5/6] refactor: remove unnecessary refs --- js/src/hooks/use-channel.ts | 12 +++--------- 1 file changed, 3 insertions(+), 9 deletions(-) diff --git a/js/src/hooks/use-channel.ts b/js/src/hooks/use-channel.ts index 0cc1fd7..2971ed4 100644 --- a/js/src/hooks/use-channel.ts +++ b/js/src/hooks/use-channel.ts @@ -22,12 +22,6 @@ export function useChannel(channelId: string | null): UseChannelResult { const [presence, setPresence] = useState([]); const [messages, setMessages] = useState([]); - // Keep a ref to avoid re-subscribing when sendMessage changes - const clientRef = useRef(client); - const channelRef = useRef(channelId); - clientRef.current = client; - channelRef.current = channelId; - useEffect(() => { if (!client || !channelId) { setPresence([]); @@ -80,11 +74,11 @@ export function useChannel(channelId: string | null): UseChannelResult { const sendMessage = useCallback( (payload: unknown) => { - if (clientRef.current && channelRef.current) { - clientRef.current.sendMessage(channelRef.current, payload); + if (client && channelId) { + client?.sendMessage(channelId, payload); } }, - [], + [client, channelId], ); return { presence, messages, sendMessage }; -- 2.54.0 From 1ffca8e2398324a7594d00b4742d71df999a140b Mon Sep 17 00:00:00 2001 From: talksik Date: Sun, 12 Apr 2026 09:58:41 -0700 Subject: [PATCH 6/6] fix: presence unreliable within same pod It was broadcasting presence changes to other pods, but not notifying clients connected to the same pod --- go/internal/pusher/hub.go | 31 +++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/go/internal/pusher/hub.go b/go/internal/pusher/hub.go index 480e019..1de54e4 100644 --- a/go/internal/pusher/hub.go +++ b/go/internal/pusher/hub.go @@ -101,6 +101,9 @@ func (h *Hub) handleSubscribe(ctx context.Context, req *subscribeRequest) { h.channels[req.channelID] = ch } + // Capture before addMember so multi-tab joins don't emit a spurious join. + wasPresentLocally := ch.hasHumanID(req.conn.humanID) + // Add to local channel ch.addMember(req.conn, req.conn.humanID) @@ -124,6 +127,16 @@ func (h *Hub) handleSubscribe(ctx context.Context, req *subscribeRequest) { Channel: req.channelID, Presence: presence, }) + + // Notify other local members. The Redis self-filter drops our own echo, + // so same-pod peers would otherwise never hear about this join. + if !wasPresentLocally { + ch.broadcast(ServerMessage{ + Type: TypeJoin, + Channel: req.channelID, + HumanID: req.conn.humanID, + }, req.conn) + } } func (h *Hub) handleUnsubscribe(ctx context.Context, req *unsubscribeRequest) { @@ -144,6 +157,16 @@ func (h *Hub) handleUnsubscribe(ctx context.Context, req *unsubscribeRequest) { slog.Error("redis unsubscribe failed", "channelId", req.channelID, "error", err) } + // Notify other local members iff the humanID is fully gone from this pod + // (multi-tab: other conns keep them present, so no leave fires). + if !ch.hasHumanID(req.conn.humanID) { + ch.broadcast(ServerMessage{ + Type: TypeLeave, + Channel: req.channelID, + HumanID: req.conn.humanID, + }, req.conn) + } + // Clean up empty local channel if ch.isEmpty() { delete(h.channels, req.channelID) @@ -192,6 +215,14 @@ func (h *Hub) handleDisconnect(ctx context.Context, conn *Conn) { slog.Error("redis unsubscribe on disconnect failed", "channelId", channelID, "error", err) } + if !ch.hasHumanID(conn.humanID) { + ch.broadcast(ServerMessage{ + Type: TypeLeave, + Channel: channelID, + HumanID: conn.humanID, + }, conn) + } + if ch.isEmpty() { delete(h.channels, channelID) } -- 2.54.0