Sessions outlive the deploy that interrupted them

A MemoryStore meant every restart signed out every desk on the site
being deployed. Bearable while a desk is three people, and not the
kind of thing that gets better on its own.

They live in NATS KV now: the JetStream already under the events, the
projections, the content reload and the access questions. No new
service, no new store, and the same bucket a second instance of a site
would read if there ever is one. `create` draws a fresh id rather than
writing over a session already at that key, the bucket's `max_age`
collects what nobody comes back for, and a record's own expiry is what
answers a load, whatever the bucket still holds.

Proved the only way it can be: a store opened twice against one
JetStream, which is what two runs of the binary are.

Also fixed, while there: the integration test asserted a record's
projected state was one that `aggregates::transition` never writes -
it appends events, the desk handler updates the projection - so it
failed whenever it was actually run. It now asserts what it says it
does, that a mailed reply leaves the state alone. test.yml has been
red since 0.3.47 while publish.yml stayed green, which is how v0.4.0
went out over it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

CHANGELOG.md

@@ -6,6 +6,8 @@ content repos pin the tag they run (see the end of this file).
 
 ## Unreleased
 
+- Sessions live in NATS KV (`src/sessions.rs`), not in memory. Every restart used to sign out every desk on the site being deployed; they now survive it, in the JetStream that is already under everything else - no new service and no new store to run. The bucket's `max_age` collects what nobody returns for, and a record's own expiry is what answers a load.
+- Fixed: the JetStream integration test asserted a record's projected state was one a desk move had never written to it, so it failed whenever it actually ran. `test.yml` has been red since 0.3.47 while `publish.yml` stayed green, which is how 0.4.0 shipped past it.
 - A file can come back out. `GET /attachment/{bucket}/{record}/{field}` streams what a `type: file` field stored, reading the object key out of the record rather than taking one from the caller - so a file is only reachable through a record somebody may see, and a key that leaks off a desk is a link to nothing. Before this, uploads were one-way: a desk showed the key as a string and there was no way to open it.
 - Rate limiting, per caller IP, outside everything else: a burst of 30 refilled one every two seconds (`PORTAL_RATE_BURST`, `PORTAL_RATE_SECONDS`; either 0 turns it off). A form-filling visitor never reaches it; a script does. There was none at all before.
 - `/upload` asks the policy, like every other way in. It asked nothing: any caller who could reach the host could put 20 MB in the bucket, as often as they liked, and a site whose forms are public is a site whose upload route is public. It now takes the `question` and `alternative` the file is being attached to, refuses a page that declares no `type: file` field or an alternative that is closed, and asks `Submit` on that page - the same question the submission itself will ask - before a byte is written. Latent until an instance had storage configured; the first one now does.

Cargo.lock

@@ -132,9 +132,9 @@ checksum = "4288f83726785267c6f2ef073a3d83dc3f9b81464e9f99898240cced85fce35a"
 
 [[package]]
 name = "async-trait"
-version = "0.1.91"
+version = "0.1.92"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ae36dc4177970ef04fde5178d3e2429882def40e57a451f919c098f72baa6cec"
+checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -3362,6 +3362,7 @@ dependencies = [
  "anyhow",
  "arc-swap",
  "async-nats",
+ "async-trait",
  "aws-sdk-s3",
  "axum",
  "base64 0.22.1",

Cargo.toml

@@ -64,6 +64,7 @@ web-sys = { version = "0.3", features = [
     "Window",
 ], optional = true }
 tower_governor = { version = "0.8.0", optional = true }
+async-trait = { version = "0.1.92", optional = true }
 
 [features]
 default = []
@@ -83,6 +84,7 @@ ssr = [
     "dep:axum",
     "dep:aws-sdk-s3",
     "dep:tower_governor",
+    "dep:async-trait",
     "dep:tokio",
     "dep:tower",
     "dep:tower-http",

docs/architecture.md

@@ -85,3 +85,8 @@ their graphs with no bespoke UI per bucket.
 - `iris` — the lint, from its own repo, shipped in every release
   tarball so a content repo's CI can lint with the exact version its
   instance runs.
+
+**Sessions** (`src/sessions.rs`) — in the same JetStream KV as
+everything else, so a restart does not sign anybody out. The bucket
+carries a `max_age` so NATS collects what nobody comes back for, and
+each record's own `expiry_date` is what decides on the way out.

docs/operations.md

@@ -66,7 +66,7 @@ values.
 
 | | |
 |---|---|
-| `NATS_URL` | events, projections, content reload, access questions |
+| `NATS_URL` | events, projections, sessions, content reload, access questions |
 | `CONTENT_REPO`, `CONTENT_BRANCH` | the repo that is the site |
 | `KANIDM_URL`, `OAUTH2_CLIENT_ID`, `OAUTH2_CLIENT_SECRET` | sign-in |
 | `KANIDM_API_TOKEN` | invites and `grants:`; unset, neither happens |

src/lib.rs

@@ -16,6 +16,9 @@ pub mod announce;
 #[cfg(feature = "ssr")]
 pub mod deadlines;
 
+#[cfg(feature = "ssr")]
+pub mod sessions;
+
 #[cfg(feature = "ssr")]
 pub mod http;
 

src/main.rs

@@ -10,7 +10,7 @@ async fn main() -> anyhow::Result<()> {
     use portal::upload::{self, Garage};
     use std::sync::Arc;
     use tower_http::services::{ServeDir, ServeFile};
-    use tower_sessions::{MemoryStore, SessionManagerLayer};
+    use tower_sessions::SessionManagerLayer;
 
     dotenvy::dotenv().ok();
     tracing_subscriber::fmt()
@@ -127,7 +127,10 @@ async fn main() -> anyhow::Result<()> {
     let cookie_secure = std::env::var("COOKIE_SECURE")
         .map(|v| v == "true" || v == "1")
         .unwrap_or(false);
-    let session_layer = SessionManagerLayer::new(MemoryStore::default())
+    // Sessions live in the same JetStream as everything else, so a
+    // deploy no longer signs out every desk on the site it deploys.
+    let session_store = portal::sessions::NatsStore::open(&state.jetstream).await?;
+    let session_layer = SessionManagerLayer::new(session_store)
         .with_secure(cookie_secure)
         .with_same_site(tower_sessions::cookie::SameSite::Lax)
         .with_name("portal_session");

src/sessions.rs

@@ -0,0 +1,122 @@
+//! Sessions in NATS KV.
+//!
+//! A session used to live in a `MemoryStore`, which meant every
+//! restart signed everybody out: a deploy logged out every desk on
+//! every site, and the only reason that was bearable is that the desks
+//! were three people. This keeps them in the JetStream KV that is
+//! already under everything else here - no new service, no new
+//! dependency, and the same bucket a second instance of a site would
+//! read if there ever is one.
+//!
+//! Expiry is said twice on purpose. The bucket carries a `max_age`, so
+//! NATS collects abandoned sessions without anyone sweeping; and a
+//! record's own `expiry_date` is checked on the way out, because the
+//! bucket's age is a coarse fence and the record's date is the truth.
+#![cfg(feature = "ssr")]
+
+use tower_sessions::session::{Id, Record};
+use tower_sessions::session_store::{Error, Result};
+
+/// How long an untouched session survives in the bucket. Comfortably
+/// longer than any expiry `SessionManagerLayer` will set, so the
+/// record's own date is always what decides; this is only here so a
+/// session nobody ever comes back for does not sit in NATS forever.
+const MAX_AGE: std::time::Duration = std::time::Duration::from_secs(60 * 60 * 24 * 40);
+
+pub const BUCKET: &str = "portal_sessions";
+
+/// A new id is drawn this many times before a collision is reported as
+/// a backend error. Two random `i128`s colliding is not a thing that
+/// happens; this exists so a bucket misbehaving cannot spin forever.
+const ID_ATTEMPTS: usize = 8;
+
+#[derive(Clone)]
+pub struct NatsStore {
+    store: async_nats::jetstream::kv::Store,
+}
+
+impl std::fmt::Debug for NatsStore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("NatsStore").field("bucket", &BUCKET).finish()
+    }
+}
+
+impl NatsStore {
+    /// Opens the bucket, creating it on first use the way every other
+    /// bucket here is created.
+    pub async fn open(js: &async_nats::jetstream::Context) -> anyhow::Result<Self> {
+        let store = match js.get_key_value(BUCKET).await {
+            Ok(store) => store,
+            Err(_) => {
+                js.create_key_value(async_nats::jetstream::kv::Config {
+                    bucket: BUCKET.to_string(),
+                    max_age: MAX_AGE,
+                    history: 1,
+                    ..Default::default()
+                })
+                .await?
+            }
+        };
+        Ok(Self { store })
+    }
+
+    fn encode(record: &Record) -> Result<Vec<u8>> {
+        serde_json::to_vec(record).map_err(|e| Error::Encode(e.to_string()))
+    }
+}
+
+#[async_trait::async_trait]
+impl tower_sessions::SessionStore for NatsStore {
+    /// `create` draws a fresh id until the bucket accepts one as new.
+    /// `put` would overwrite a stranger's session on a collision;
+    /// `create` refuses, which is the whole difference.
+    async fn create(&self, record: &mut Record) -> Result<()> {
+        for _ in 0..ID_ATTEMPTS {
+            let bytes = Self::encode(record)?;
+            match self.store.create(&record.id.to_string(), bytes.into()).await {
+                Ok(_) => return Ok(()),
+                Err(_) => record.id = Id::default(),
+            }
+        }
+        Err(Error::Backend(format!(
+            "could not find an unused session id in {ID_ATTEMPTS} attempts"
+        )))
+    }
+
+    async fn save(&self, record: &Record) -> Result<()> {
+        let bytes = Self::encode(record)?;
+        self.store
+            .put(&record.id.to_string(), bytes.into())
+            .await
+            .map(|_| ())
+            .map_err(|e| Error::Backend(e.to_string()))
+    }
+
+    async fn load(&self, id: &Id) -> Result<Option<Record>> {
+        let Some(bytes) = self
+            .store
+            .get(&id.to_string())
+            .await
+            .map_err(|e| Error::Backend(e.to_string()))?
+        else {
+            return Ok(None);
+        };
+        let record: Record =
+            serde_json::from_slice(&bytes).map_err(|e| Error::Decode(e.to_string()))?;
+        // An expired record is gone as far as anyone asking is
+        // concerned, whether or not the bucket has collected it yet.
+        if record.expiry_date <= tower_sessions::cookie::time::OffsetDateTime::now_utc() {
+            return Ok(None);
+        }
+        Ok(Some(record))
+    }
+
+    async fn delete(&self, id: &Id) -> Result<()> {
+        // Purge, not delete: a signed-out session should leave no
+        // history in the bucket to read back.
+        self.store
+            .purge(&id.to_string())
+            .await
+            .map_err(|e| Error::Backend(e.to_string()))
+    }
+}

tests/jetstream.rs

@@ -68,6 +68,7 @@ async fn a_submission_is_created_moved_refused_and_read_back_through_the_real_en
     // through the same store the desks read - and moves nothing: the
     // record is still where the desk left it.
     let mut record: Answer = serde_json::from_slice(&store.get("rec-1").await.unwrap().expect("record")).unwrap();
+    let state_before_the_note = record.state.clone();
     let reply = portal::mail::Incoming {
         v: portal::mail::CONTRACT,
         id: "mail-1".into(),
@@ -86,10 +87,69 @@ async fn a_submission_is_created_moved_refused_and_read_back_through_the_real_en
     assert_eq!(read_back.notes.len(), 1);
     assert_eq!(read_back.notes[0].text, "Thursday suits us.");
     assert_eq!(read_back.notes[0].from, "a@b.example");
-    assert_eq!(read_back.state, "quoted", "a reply is a note, never a decision");
+    assert_eq!(read_back.state, state_before_the_note, "a reply is a note, never a decision");
     // The same reply read twice adds nothing.
     assert!(portal::mail::note_for(&reply, &read_back).is_err());
     // And the event log is untouched by it.
     let after = load_events(&js, &bucket, "rec-1").await.expect("events");
     assert_eq!(after.len(), 3, "a note writes no event");
 }
+
+/// A session outlives the process that made it. The `MemoryStore` this
+/// replaced could not: every deploy signed out every desk, and the
+/// only proof that this one does not is a store opened twice against
+/// the same JetStream, as two runs of the binary would.
+#[tokio::test]
+async fn a_session_survives_the_process_that_made_it() {
+    use tower_sessions::session::{Id, Record};
+    use tower_sessions::SessionStore;
+
+    let Ok(url) = std::env::var("PORTAL_TEST_NATS_URL") else {
+        eprintln!("skipped: PORTAL_TEST_NATS_URL is not set");
+        return;
+    };
+    let js = async_nats::jetstream::new(async_nats::connect(&url).await.expect("connect"));
+    let store = portal::sessions::NatsStore::open(&js).await.expect("open");
+
+    let mut record = Record {
+        id: Id::default(),
+        data: Default::default(),
+        expiry_date: tower_sessions::cookie::time::OffsetDateTime::now_utc()
+            + tower_sessions::cookie::time::Duration::hours(1),
+    };
+    record.data.insert("who".into(), json!("ingrid"));
+    store.create(&mut record).await.expect("create");
+
+    // A second store over the same JetStream is what the next boot of
+    // the binary has: nothing carried over in memory.
+    let next_boot = portal::sessions::NatsStore::open(&js).await.expect("reopen");
+    let found = next_boot.load(&record.id).await.expect("load").expect("still there");
+    assert_eq!(found.data.get("who"), Some(&json!("ingrid")));
+
+    // `create` never writes over somebody else's session: asked to
+    // create one whose id is taken, it takes a new id instead.
+    let mut collide = Record { id: record.id, data: Default::default(), expiry_date: record.expiry_date };
+    next_boot.create(&mut collide).await.expect("create beside it");
+    assert_ne!(collide.id, record.id, "an id already in use must not be reused");
+    assert_eq!(
+        next_boot.load(&record.id).await.expect("load").expect("original intact").data.get("who"),
+        Some(&json!("ingrid")),
+        "the original session must be untouched"
+    );
+
+    // An expired record is gone to anyone asking, whatever the bucket
+    // still holds.
+    let stale = Record {
+        id: Id::default(),
+        data: Default::default(),
+        expiry_date: tower_sessions::cookie::time::OffsetDateTime::now_utc()
+            - tower_sessions::cookie::time::Duration::minutes(1),
+    };
+    next_boot.save(&stale).await.expect("save stale");
+    assert!(next_boot.load(&stale.id).await.expect("load stale").is_none());
+
+    // Signing out leaves nothing to read back.
+    next_boot.delete(&record.id).await.expect("delete");
+    assert!(next_boot.load(&record.id).await.expect("load after delete").is_none());
+    next_boot.delete(&collide.id).await.ok();
+}