From adbcabc93f221dd677ffe3b2dddd230c35d2daa7 Mon Sep 17 00:00:00 2001
From: "nekomimi.pet" <meowskulls@nekomimi.pet>
Date: Mon, 26 Oct 2026 01:59:57 -0400
Subject: [PATCH] sessions stream: open the collection follow on the stream's
 first wake, so listing thousands of sessions no longer outlasts mist's 1 s
 chunked start and ends the stream empty

---
 .../daemon/http_session_collection.gleam      | 26 ++++----
 test/e2e/session_status_test.py               | 59 ++++++++++++++++++-
 2 files changed, 74 insertions(+), 11 deletions(-)

diff --git a/src/albedo/daemon/http_session_collection.gleam b/src/albedo/daemon/http_session_collection.gleam
index 4680bd49..eda3292a 100644
--- a/src/albedo/daemon/http_session_collection.gleam
+++ b/src/albedo/daemon/http_session_collection.gleam
@@ -157,14 +157,15 @@ pub fn read(
           live,
           http_stream.response(live, coded),
           fn(subject) {
-            let follow =
-              open(config, host, parameters, fn() {
-                process.send(subject, StreamWake)
-              })
             process.send(subject, StreamWake)
             let _ = clock.send_after(subject, 5000, StreamTick)
             CollectionStream(
-              follow,
+              None,
+              fn() {
+                open(config, host, parameters, fn() {
+                  process.send(subject, StreamWake)
+                })
+              },
               subject,
               http_stream.writer(live, coded),
               True,
@@ -275,7 +276,10 @@ type StreamMessage {
 
 type CollectionStream {
   CollectionStream(
-    follow: Result(Follow, http_api.Failure),
+    /// None until the first message: opening reads every session in scope,
+    /// which can outlast the second mist gives a chunked response to start.
+    follow: Option(Result(Follow, http_api.Failure)),
+    open: fn() -> Result(Follow, http_api.Failure),
     subject: Subject(StreamMessage),
     writer: http_stream.Writer,
     initial: Bool,
@@ -429,13 +433,15 @@ fn collection_stream_loop(
   message: StreamMessage,
   _connection: mist.Connection,
 ) -> mist.ChunkNext(CollectionStream) {
+  let opened = option.lazy_unwrap(state.follow, state.open)
+  let state = CollectionStream(..state, follow: Some(opened))
   let batch = {
-    use follow <- result.try(state.follow)
+    use follow <- result.try(opened)
     next(follow, state.initial)
   }
   case batch {
     Error(_) -> {
-      let _ = result.map(state.follow, close)
+      let _ = result.map(opened, close)
       collection_overflow(state.writer)
     }
     Ok(batch) -> {
@@ -446,14 +452,14 @@ fn collection_stream_loop(
       }
       case sent {
         Error(_) -> {
-          let _ = result.map(state.follow, close)
+          let _ = result.map(opened, close)
           mist.ChunkStop
         }
         Ok(_) -> {
           case state.initial {
             True -> process.send(state.subject, StreamWake)
             False -> {
-              let _ = result.map(state.follow, ack(_, message == StreamWake))
+              let _ = result.map(opened, ack(_, message == StreamWake))
               case message {
                 StreamWake -> Nil
                 StreamTick -> {
diff --git a/test/e2e/session_status_test.py b/test/e2e/session_status_test.py
index 86f9c3e6..d414766f 100644
--- a/test/e2e/session_status_test.py
+++ b/test/e2e/session_status_test.py
@@ -3,13 +3,15 @@
 import http.client
 import json
 import queue
+import sqlite3
 import threading
 import time
 import unittest
 import urllib.error
 import urllib.parse
+import uuid
 
-from harness import Albedo, Provider, latest_user, python, text
+from harness import Albedo, Provider, exclusive, latest_user, python, text
 
 # Long enough for the daemon's sidebar pace plus a loaded test host.
 PUSH_SECONDS = 25
@@ -180,3 +182,58 @@ class SessionStatusPushTests(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+# exclusive: fills the store with thousands of sessions
+@exclusive
+class CollectionStreamTests(unittest.TestCase):
+    def test_the_sessions_stream_starts_however_many_sessions_it_lists(self):
+        # Following the collection reads every session it covers. That once ran
+        # where mist gives a chunked response one second to start, and a big
+        # store (or a busy host) answered every watcher with an empty stream.
+        with Albedo() as app:
+            seed = app.session()
+
+            def fill(app):
+                with sqlite3.connect(app.home / "albedo.sqlite") as db:
+                    columns = [
+                        row[1] for row in db.execute("PRAGMA table_info(sessions)")
+                    ]
+                    row = db.execute(
+                        f"SELECT {','.join(columns)} FROM sessions WHERE id=?", (seed,)
+                    ).fetchone()
+                    at = columns.index("id")
+                    db.executemany(
+                        f"INSERT INTO sessions({','.join(columns)}) "
+                        f"VALUES({','.join('?' * len(columns))})",
+                        (
+                            [*row[:at], str(uuid.uuid4()), *row[at + 1 :]]
+                            for _ in range(8000)
+                        ),
+                    )
+
+            app.restart(prepare=fill)
+            connection = http.client.HTTPConnection(
+                "127.0.0.1", app.connection["port"], timeout=60
+            )
+            self.addCleanup(connection.close)
+            connection.request(
+                "GET",
+                "/sessions?scope=all",
+                headers={
+                    "Authorization": "Bearer " + app.connection["token"],
+                    "Accept": "text/event-stream",
+                },
+            )
+            first = next(
+                (
+                    line
+                    for line in connection.getresponse()
+                    if line.startswith(b"data: ")
+                ),
+                None,
+            )
+            if first is None:
+                self.fail("the sessions stream ended without a batch")
+            events = [event["type"] for event in json.loads(first[6:])["events"]]
+            self.assertEqual(events[:2], ["ready", "reset"])
-- 
knotmirror 0.0.1