From 08a2ed94a309919d38154616956f571bf7765f23 Mon Sep 17 00:00:00 2001 From: dhkatz Date: Fri, 30 Oct 2026 05:37:53 -0700 Subject: [PATCH] Keep the gateway connection alive and resume hosted queues after reconnecting The client doesn't ping while it's sending, so when the proxy answers its frames itself (everything it sends in a hosted room), IMVU's gateway saw an idle connection and closed it after about a minute. A ping now goes to the gateway in place of each answered frame. And after a reconnect, hosted rooms' queues are taken out of C2gOpenFloodgates and answered here (the gateway never answered them, so the client gave up and closed the room). Co-Authored-By: Claude Opus 5.5 --- internal/imq/frame_test.go | 14 +++++++++ internal/imq/messages.go | 59 ++++++++++++++++++++++++++++++------ internal/proxy/connect.go | 30 ++++++++++++++++-- internal/proxy/proxy_test.go | 3 +- 4 files changed, 94 insertions(+), 12 deletions(-) diff --git a/internal/imq/frame_test.go b/internal/imq/frame_test.go index e70f90b..f02d22c 100644 --- a/internal/imq/frame_test.go +++ b/internal/imq/frame_test.go @@ -97,3 +97,17 @@ func TestSubscriptionOps(t *testing.T) { t.Fatalf("all %v", all) } } + +func TestFloodgates(t *testing.T) { + state := func(q string, seq, op uint64) []byte { + return appendVarintField(appendVarintField(appendBytesField(nil, 1, []byte(q)), 2, seq), 3, op) + } + data := appendBytesField(appendBytesField(nil, 1, state("/user/1", 38, 19)), 1, state("/chat/2100000001", 29, 20)) + states, err := ParseFloodgates(data) + if err != nil || len(states) != 2 || states[1].Queue != "/chat/2100000001" || states[1].OpID != 20 { + t.Fatalf("%+v %v", states, err) + } + if string(MarshalFloodgates(states)) != string(data) || string(MarshalFloodgates(states[:1])) != string(appendBytesField(nil, 1, state("/user/1", 38, 19))) { + t.Fatal("re-encoded differently") + } +} diff --git a/internal/imq/messages.go b/internal/imq/messages.go index a032a48..cce4ed6 100644 --- a/internal/imq/messages.go +++ b/internal/imq/messages.go @@ -4,15 +4,17 @@ import "errors" // Message types the proxy reads or writes. const ( - C2gSendMessage = 1304 - C2gUnsubscribe = 1308 - C2gSubscribe = 1309 - G2cResult = 3102 - G2cSendMessage = 3104 - G2cStateChange = 3105 - G2cJoinedQueue = 3106 - G2cLeftQueue = 3107 - G2cCreateMount = 3108 + C2gPing = 1301 + C2gSendMessage = 1304 + C2gOpenFloodgates = 1306 + C2gUnsubscribe = 1308 + C2gSubscribe = 1309 + G2cResult = 3102 + G2cSendMessage = 3104 + G2cStateChange = 3105 + G2cJoinedQueue = 3106 + G2cLeftQueue = 3107 + G2cCreateMount = 3108 ) // Mount types (G2cCreateMount). @@ -314,3 +316,42 @@ func StateChange(opID uint64, userID, queue, mount string, props []Property, seq b = appendBytesField(appendBytesField(appendBytesField(b, 2, []byte(userID)), 3, []byte(queue)), 4, []byte(mount)) return appendVarintField(appendProperties(b, 5, props), 6, sequence) } + +// ParseFloodgates reads a C2gOpenFloodgates's data (after reconnecting, the queues to resume): each queue's state +// as a QueueOp (its op ID for a result), keeping the entry as sent. +func ParseFloodgates(data []byte) ([]QueueOp, error) { + fs, err := fields(data) + if err != nil { + return nil, err + } + var out []QueueOp + for _, f := range fs { + if f.num != 1 || !f.isLen { + return nil, errors.New("imq: unexpected field in C2gOpenFloodgates") + } + sub, err := fields(f.bytes) + if err != nil { + return nil, err + } + q := QueueOp{raw: f.bytes} + for _, sf := range sub { + switch { + case sf.num == 1 && sf.isLen: + q.Queue = string(sf.bytes) + case sf.num == 3 && !sf.isLen: + q.OpID = sf.varint + } + } + out = append(out, q) + } + return out, nil +} + +// MarshalFloodgates encodes a C2gOpenFloodgates's data from queue states read by ParseFloodgates. +func MarshalFloodgates(states []QueueOp) []byte { + var b []byte + for _, q := range states { + b = appendBytesField(b, 1, q.raw) + } + return b +} diff --git a/internal/proxy/connect.go b/internal/proxy/connect.go index 9f7e45d..e576b52 100644 --- a/internal/proxy/connect.go +++ b/internal/proxy/connect.go @@ -253,12 +253,33 @@ func (st *stream) filtered(f imq.Frame, frame []byte) []byte { } rest := router.Route(st.conn, f.Type == imq.C2gUnsubscribe, sub) if rest.Empty() { - return nil + return keepalive() } if len(rest.All()) == len(sub.All()) { return frame } return imq.EncodeFrame(f.Type, rest.Marshal()) + case imq.C2gOpenFloodgates: + // Resuming queues after a reconnect: the router's are answered here, the rest go to the gateway. + router, ok := st.watcher.(IMQRouter) + if !ok { + return frame + } + states, err := imq.ParseFloodgates(f.Data) + if err != nil { + return frame + } + var rest []imq.QueueOp + for _, q := range states { + if router.Route(st.conn, false, imq.Subscription{WithResults: []imq.QueueOp{q}}).Empty() { + continue + } + rest = append(rest, q) + } + if len(rest) == len(states) { + return frame + } + return imq.EncodeFrame(imq.C2gOpenFloodgates, imq.MarshalFloodgates(rest)) case imq.G2cSendMessage: m, err := imq.ParseDelivered(f.Data) if err != nil { @@ -284,7 +305,7 @@ func (st *stream) filtered(f imq.Frame, frame []byte) []byte { if st.reply != nil { st.reply.Write(imq.EncodeFrame(imq.G2cResult, imq.Result(m.OpID, 0))) } - return nil + return keepalive() } if bytes.Equal(out, m.Message) { return frame @@ -293,6 +314,11 @@ func (st *stream) filtered(f imq.Frame, frame []byte) []byte { return imq.EncodeFrame(imq.C2gSendMessage, m.Marshal()) } +// keepalive is what goes to the gateway in place of a frame the proxy answered itself: a ping. The client doesn't +// ping while it's sending, so without it the gateway sees an idle connection and closes it (in a hosted room, where +// everything the client sends is answered here). +func keepalive() []byte { return imq.EncodeFrame(imq.C2gPing, nil) } + func (st *stream) summary() string { if st.other { return fmt.Sprintf("%s %dB", st.name, st.bytes) diff --git a/internal/proxy/proxy_test.go b/internal/proxy/proxy_test.go index aba029d..4fa11d6 100644 --- a/internal/proxy/proxy_test.go +++ b/internal/proxy/proxy_test.go @@ -427,7 +427,8 @@ func TestConnectFiltersOutgoingMessages(t *testing.T) { } conn.Close() got := <-received - want := append(send(1, "messages", "hi!"), send(3, "control", "x")...) + // The dropped message's place: a ping, so the gateway doesn't see an idle connection. + want := append(append(send(1, "messages", "hi!"), imq.EncodeFrame(imq.C2gPing, nil)...), send(3, "control", "x")...) if !bytes.Equal(got, want) { t.Fatalf("gateway got %q, want %q", got, want) }