Redis is not Unsubscribed when the Client Disconnects

[genematx]

The Problem

When receiving updates via WebSockets, the server process subscribes to Redis (on behalf of the client); e.g. in Redis logs this can be observed as:

1773444651.297893 [0 192.168.65.1:57132] "SUBSCRIBE" "notify:1"
1773444651.299970 [0 192.168.65.1:22509] "CLIENT" "SETINFO" "LIB-NAME" "redis-py"
1773444651.299979 [0 192.168.65.1:22509] "CLIENT" "SETINFO" "LIB-VER" "7.0.1"
1773444651.300702 [0 192.168.65.1:22509] "GET" "sequence:1"
...

In the end of the client session, when the client disconnects, an unsubscribe even should be posted. This is the seemingly the case with the TTLCacheDatastore / _TestClientWebsocketWrapper used in tests.

...
1773444651.549106 [0 192.168.65.1:28832] "HMGET" "data:1:7" "payload" "metadata"
1773444653.663609 [0 192.168.65.1:57132] "UNSUBSCRIBE" "notify:1"

However, with the usual (production) RedisStreamingDatastore / _RegularWebsocketWrapper, the UNSUBSCRIBE event doesn't happen after the client is disconnected. Only after the server is turned off, Redis receives all UNSUBSCRIBE events at once.

A Way to Reproduce

1. Start Redis and Postgres (possibly can use SQLite too) in containers, e.g. use the following docker commands:

docker run -p 5432:5432 -e POSTGRES_PASSWORD=secret -d docker.io/postgres:16
docker run --name tiled-redis -p 6379:6379 -d redis

2. Start a local Tiled server with the following config file:

authentication:
  single_user_api_key: "secret"
  allow_anonymous_access: true
trees:
  - path: /
    tree: catalog
    args:
      uri: postgresql://tiled:secret@localhost:5432/catalog
      init_if_not_exists: true
streaming_cache:
  uri: "redis://localhost:6379"

3. Exec into the Redis container:

docker exec -it tiled-redis redis-cli

and start monitoring the log:

127.0.0.1:6379> monitor
OK

4. Run the following client-side script:

client = from_uri("http://localhost:8000", api_key="secret")['test/ws_test']

for i in range(5):
    sub = x.subscribe()
    sub.start_in_thread()
    sub.child_created.add_callback(lambda : _)

    time.sleep(1)
    sub.disconnect()

5. When the above script is run repeatedly, Redis registers the following messages:

1773665801.420791 [0 192.168.65.1:52035] "INCRBY" "sequence:19" "1"
1773665801.421712 [0 192.168.65.1:52035] "MULTI"
1773665801.421740 [0 192.168.65.1:52035] "HSET" "data:19:15" "sequence" "15" "metadata" "{\"type\":\"container-child-created\",\"sequence\":15,\"timestamp\":\"2026-03-16T08:56:41.422191\",\"key\":\"63927592-1fcf-46e2-8309-651dc46714b1\",\"structure_family\":\"container\",\"specs\":[],\"metadata\":{},\"data_sources\":[],\"access_blob\":{}}"
1773665801.421759 [0 192.168.65.1:52035] "EXPIRE" "data:19:15" "3600"
1773665801.421764 [0 192.168.65.1:52035] "PUBLISH" "notify:19" "15"
1773665801.421766 [0 192.168.65.1:52035] "EXPIRE" "sequence:19" "2592000"
1773665801.421769 [0 192.168.65.1:52035] "EXEC"
1773665801.443923 [0 192.168.65.1:52035] "SUBSCRIBE" "notify:1937"
1773665802.479654 [0 192.168.65.1:40848] "CLIENT" "SETINFO" "LIB-NAME" "redis-py"
1773665802.479673 [0 192.168.65.1:40848] "CLIENT" "SETINFO" "LIB-VER" "7.2.1"
1773665802.480731 [0 192.168.65.1:40848] "SUBSCRIBE" "notify:1937"
1773665803.519538 [0 192.168.65.1:24100] "CLIENT" "SETINFO" "LIB-NAME" "redis-py"
1773665803.519565 [0 192.168.65.1:24100] "CLIENT" "SETINFO" "LIB-VER" "7.2.1"
1773665803.520780 [0 192.168.65.1:24100] "SUBSCRIBE" "notify:1937"
1773665804.565593 [0 192.168.65.1:45566] "CLIENT" "SETINFO" "LIB-NAME" "redis-py"
1773665804.565620 [0 192.168.65.1:45566] "CLIENT" "SETINFO" "LIB-VER" "7.2.1"
1773665804.566795 [0 192.168.65.1:45566] "SUBSCRIBE" "notify:1937"
1773665805.607610 [0 192.168.65.1:53475] "CLIENT" "SETINFO" "LIB-NAME" "redis-py"
1773665805.607636 [0 192.168.65.1:53475] "CLIENT" "SETINFO" "LIB-VER" "7.2.1"
1773665805.608770 [0 192.168.65.1:53475] "SUBSCRIBE" "notify:1937"

while the sever logs show that the connection has been closed:

[19eb8f945d002b8c] INFO:     127.0.0.1:50554 - "WebSocket /api/v1/stream/single/test//ws_test//63927592-1fcf-46e2-8309-651dc46714b1?envelope_format=msgpack" [accepted]
[-] INFO:     connection open
[-] INFO:     connection closed
[2e3ff2bfe9a9641e] INFO:     127.0.0.1:50556 - "WebSocket /api/v1/stream/single/test//ws_test//63927592-1fcf-46e2-8309-651dc46714b1?envelope_format=msgpack" [accepted]
[-] INFO:     connection open
[-] INFO:     connection closed
[a3d304e3f4566714] INFO:     127.0.0.1:50559 - "WebSocket /api/v1/stream/single/test//ws_test//63927592-1fcf-46e2-8309-651dc46714b1?envelope_format=msgpack" [accepted]
[-] INFO:     connection open
[-] INFO:     connection closed
[a7daeb81ed078b19] INFO:     127.0.0.1:50562 - "WebSocket /api/v1/stream/single/test//ws_test//63927592-1fcf-46e2-8309-651dc46714b1?envelope_format=msgpack" [accepted]
[-] INFO:     connection open
[-] INFO:     connection closed
[8b45dd4447434973] INFO:     127.0.0.1:50565 - "WebSocket /api/v1/stream/single/test//ws_test//63927592-1fcf-46e2-8309-651dc46714b1?envelope_format=msgpack" [accepted]
[-] INFO:     connection open
[-] INFO:     connection closed

Finally, when the server is shut down:

1773665934.269351 [0 192.168.65.1:38335] "UNSUBSCRIBE" "notify:1928"
1773665934.269372 [0 192.168.65.1:51410] "UNSUBSCRIBE" "notify:1930"
1773665934.269378 [0 192.168.65.1:24100] "UNSUBSCRIBE" "notify:1937"
1773665934.269383 [0 192.168.65.1:45566] "UNSUBSCRIBE" "notify:1937"
1773665934.269390 [0 192.168.65.1:43636] "UNSUBSCRIBE" "notify:1929"
1773665934.269394 [0 192.168.65.1:52035] "UNSUBSCRIBE" "notify:1937"
1773665934.269399 [0 192.168.65.1:40848] "UNSUBSCRIBE" "notify:1937"
1773665934.269403 [0 192.168.65.1:53475] "UNSUBSCRIBE" "notify:1937"

Note, that calling x.close_stream() on the client, does issue the UNSUBSCRIBE event.