feat(hosted): inactive-user TTL reclaim (#4561 P5) (#4577)

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: sanity

crates/core/src/client_events/websocket.rs

@@ -1089,8 +1089,13 @@ async fn websocket_commands(
     // that carries a `UserSecretContext`, which only the hosted serve path
     // (which DOES install the limiter) ever derives.
     op_rate_limiter: Option<Extension<UserOpRateLimiter>>,
+    // The node's secrets dir for stamping per-user last-activity markers
+    // (#4561). OPTIONAL: only `serve_client_api_in_impl` installs it; the
+    // standalone test paths install nothing, so absence ⇒ no activity stamping.
+    activity_secrets_dir: Option<Extension<crate::server::ActivitySecretsDir>>,
 ) -> Response {
     let op_rate_limiter = op_rate_limiter.map(|Extension(l)| l);
+    let activity_secrets_dir = activity_secrets_dir.map(|Extension(d)| d);
     let on_upgrade = move |ws: WebSocket| async move {
         // Get the data we need from the DashMap
         // Track whether a token was provided but is invalid (stale/expired)
@@ -1149,6 +1154,7 @@ async fn websocket_commands(
             auth_and_instance,
             user_context,
             op_rate_limiter,
+            activity_secrets_dir,
             token_is_invalid,
             encoding_protoc,
             api_version,
@@ -1212,6 +1218,10 @@ async fn websocket_interface(
     // user's connections; consulted in `process_client_request` only when this
     // connection carries a `user_context`.
     op_rate_limiter: Option<UserOpRateLimiter>,
+    // The node's secrets dir for per-user last-activity stamping (#4561). `None`
+    // on standalone/test paths; an empty path also disables stamping. Used only
+    // when this connection carries a `user_context` (hosted mode).
+    activity_secrets_dir: Option<crate::server::ActivitySecretsDir>,
     token_is_invalid: bool,
     encoding_protoc: EncodingProtocol,
     api_version: ApiVersion,
@@ -1220,6 +1230,16 @@ async fn websocket_interface(
     let (mut response_rx, client_id) =
         new_client_connection(&request_sender, auth_token.clone()).await?;
     let (mut server_sink, mut client_stream) = ws.split();
+
+    // Stamp last-activity on CONNECT (#4561, inactive-user TTL): a fresh
+    // connection from a hosted user is activity, so refresh their `.last_seen`
+    // marker right away (debounced — a no-op if a recent stamp exists). Gated on
+    // `user_context.is_some()` so Local/non-hosted connections never write a
+    // marker. Best-effort and off the request hot path (runs once per
+    // connection). See `stamp_activity` for the wall-clock + debounce details.
+    if let (Some(ctx), Some(dir)) = (user_context.as_ref(), activity_secrets_dir.as_ref()) {
+        stamp_activity(dir, ctx);
+    }
     let contract_updates: Arc<Mutex<VecDeque<(_, mpsc::Receiver<HostResult>)>>> =
         Arc::new(Mutex::new(VecDeque::new()));
 
@@ -1336,6 +1356,7 @@ async fn websocket_interface(
                     auth_token.as_mut().map(|t| t.1),
                     user_context.as_ref(),
                     op_rate_limiter.as_ref(),
+                    activity_secrets_dir.as_ref(),
                     api_version,
                     &mut delegate_rate_limiter,
                     &mut conn_state,
@@ -1614,6 +1635,28 @@ fn extract_stream_content(
 /// (control / reassembly, already bounded per-connection by the stdlib) — does
 /// not count, so a client can always authenticate and disconnect even while
 /// throttled.
+/// Stamp a hosted user's durable last-activity marker (#4561, P5 of #4381,
+/// inactive-user TTL). Wraps [`crate::wasm_runtime::stamp_user_last_seen`] with
+/// the real wall-clock now and the default debounce, and short-circuits when no
+/// secrets dir is configured (empty path ⇒ standalone/test composition with no
+/// tree to mark). Cheap on the hot path: a single `stat`, and a write only when
+/// the marker is older than the debounce interval.
+fn stamp_activity(dir: &crate::server::ActivitySecretsDir, ctx: &UserSecretContext) {
+    let base = dir.0.as_ref();
+    if base.as_os_str().is_empty() {
+        return;
+    }
+    // Single shared wall-clock source (incl. its clamp-to-0) so the stamp hook
+    // and the reclaim sweep can never drift.
+    let now = crate::wasm_runtime::wall_clock_unix_secs();
+    crate::wasm_runtime::stamp_user_last_seen(
+        base,
+        ctx.user_id(),
+        now,
+        crate::wasm_runtime::DEFAULT_LAST_SEEN_DEBOUNCE_SECS,
+    );
+}
+
 fn is_rate_limited_op(req: &ClientRequest<'_>) -> bool {
     matches!(
         req,
@@ -1630,6 +1673,7 @@ async fn process_client_request(
     origin_contract: Option<ContractInstanceId>,
     user_context: Option<&UserSecretContext>,
     op_rate_limiter: Option<&UserOpRateLimiter>,
+    activity_secrets_dir: Option<&crate::server::ActivitySecretsDir>,
     api_version: ApiVersion,
     rate_limiter: &mut DelegateRateLimiter,
     conn_state: &mut ConnectionState,
@@ -1766,6 +1810,18 @@ async fn process_client_request(
         }
     }
 
+    // Stamp last-activity for this hosted user on EVERY request (#4561,
+    // inactive-user TTL). Done BEFORE the rate-limit check below so a user who
+    // is currently being THROTTLED still counts as active and can never be
+    // reclaimed as "abandoned" — a rate-limited request is the strongest
+    // possible evidence the user is present. Gated on `user_context.is_some()`
+    // (hosted mode honored a token), and debounced to ~free on the hot path
+    // (a single `stat`, no write, unless the marker is >1h stale). See
+    // `stamp_activity`.
+    if let (Some(ctx), Some(dir)) = (user_context, activity_secrets_dir) {
+        stamp_activity(dir, ctx);
+    }
+
     // Per-user operation rate limit (#4561, P5 of #4381). Bounds how fast a
     // single HOSTED user can issue WORK-CAUSING operations so one visitor cannot
     // flood the node's executor/network. See `is_rate_limited_op` for exactly
@@ -3817,6 +3873,7 @@ mod tests {
             None,
             user_context,
             op_rate_limiter,
+            None, // no activity stamping in rate-limit unit tests
             ApiVersion::V1,
             &mut delegate_rate_limiter,
             &mut conn_state,

crates/core/src/config.rs

@@ -134,6 +134,28 @@ pub struct ConfigArgs {
     #[arg(long = "per-user-secret-quota", env = "PER_USER_SECRET_QUOTA")]
     pub per_user_secret_quota_bytes: Option<u64>,
 
+    /// Inactivity TTL, in seconds, after which a HOSTED user's entire
+    /// per-user data is reclaimed by a background sweep (#4561, P5 of #4381).
+    /// Keeps a public "try Freenet" node a transient demo with bounded storage:
+    /// a visitor who walks away has their namespace reclaimed after this many
+    /// real-calendar seconds of inactivity (durable across restarts). Default:
+    /// 2_592_000 (30 days). `0` disables the sweep entirely. Has NO effect
+    /// outside hosted mode — Local single-user data is never enumerated or
+    /// reclaimed (it lives outside the `users/<id>/` tree the sweep touches).
+    #[arg(long = "per-user-inactive-ttl", env = "PER_USER_INACTIVE_TTL")]
+    pub per_user_inactive_ttl_secs: Option<u64>,
+
+    /// How often, in seconds, the inactive-user reclaim sweep runs (#4561).
+    /// Only relevant when hosted mode is on and `per-user-inactive-ttl` is
+    /// non-zero. Default: 3_600 (hourly) — far finer than the 30-day TTL, so
+    /// reclamation lag is negligible while keeping the sweep's disk-walk cost
+    /// trivial. Must be > 0; `0` is treated as the default.
+    #[arg(
+        long = "inactive-user-sweep-interval",
+        env = "INACTIVE_USER_SWEEP_INTERVAL"
+    )]
+    pub inactive_user_sweep_interval_secs: Option<u64>,
+
     /// Byte budget for the compiled-WASM **contract** module cache. The
     /// **delegate** cache gets a fraction of this value
     /// (`DELEGATE_MODULE_CACHE_BUDGET_DIVISOR`, currently 1/4), so the combined
@@ -206,6 +228,8 @@ impl Default for ConfigArgs {
             max_blocking_threads: None,
             max_hosting_storage: None,
             per_user_secret_quota_bytes: None,
+            per_user_inactive_ttl_secs: None,
+            inactive_user_sweep_interval_secs: None,
             module_cache_budget_bytes: None,
             shutdown_drain_secs: None,
             telemetry: Default::default(),
@@ -461,6 +485,10 @@ impl ConfigArgs {
                 .get_or_insert(cfg.max_hosting_storage);
             self.per_user_secret_quota_bytes
                 .get_or_insert(cfg.per_user_secret_quota_bytes);
+            self.per_user_inactive_ttl_secs
+                .get_or_insert(cfg.per_user_inactive_ttl_secs);
+            self.inactive_user_sweep_interval_secs
+                .get_or_insert(cfg.inactive_user_sweep_interval_secs);
             self.module_cache_budget_bytes
                 .get_or_insert(cfg.module_cache_budget_bytes);
             self.shutdown_drain_secs
@@ -899,6 +927,9 @@ impl ConfigArgs {
                     .ws_api
                     .per_user_export_min_interval_secs
                     .unwrap_or_else(default_per_user_export_min_interval_secs),
+                // Runtime-only: resolve the secrets dir for this mode so the WS
+                // serve layer can stamp per-user activity markers (#4561).
+                secrets_dir: config_paths.secrets_dir(mode),
             },
             secrets,
             log_level: self.log_level.unwrap_or(tracing::log::LevelFilter::Info),
@@ -915,6 +946,23 @@ impl ConfigArgs {
             per_user_secret_quota_bytes: self
                 .per_user_secret_quota_bytes
                 .unwrap_or(crate::wasm_runtime::DEFAULT_PER_USER_SECRET_QUOTA_BYTES as u64),
+            per_user_inactive_ttl_secs: self
+                .per_user_inactive_ttl_secs
+                .unwrap_or(default_per_user_inactive_ttl_secs()),
+            inactive_user_sweep_interval_secs: {
+                // `0` means "use the default" (an interval of 0 is meaningless —
+                // the sweep would otherwise floor it to 1s and hammer the disk).
+                // Remap here so the resolved value always reflects the documented
+                // semantics, rather than relying on a downstream `.max(1)`.
+                let v = self
+                    .inactive_user_sweep_interval_secs
+                    .unwrap_or(default_inactive_user_sweep_interval_secs());
+                if v == 0 {
+                    default_inactive_user_sweep_interval_secs()
+                } else {
+                    v
+                }
+            },
             module_cache_budget_bytes: self
                 .module_cache_budget_bytes
                 .unwrap_or_else(crate::wasm_runtime::default_module_cache_budget_bytes),
@@ -1054,6 +1102,23 @@ pub struct Config {
         rename = "per-user-secret-quota"
     )]
     pub per_user_secret_quota_bytes: u64,
+    /// Inactivity TTL in seconds after which a HOSTED user's entire per-user
+    /// data is reclaimed by a background sweep (#4561, P5 of #4381). Durable,
+    /// real-calendar time (survives restarts). Default 2_592_000 (30 days);
+    /// `0` disables the sweep. No effect outside hosted mode — Local
+    /// single-user data is never enumerated or reclaimed.
+    #[serde(
+        default = "default_per_user_inactive_ttl_secs",
+        rename = "per-user-inactive-ttl"
+    )]
+    pub per_user_inactive_ttl_secs: u64,
+    /// How often (seconds) the inactive-user reclaim sweep runs (#4561). Only
+    /// relevant in hosted mode with a non-zero TTL. Default 3_600 (hourly).
+    #[serde(
+        default = "default_inactive_user_sweep_interval_secs",
+        rename = "inactive-user-sweep-interval"
+    )]
+    pub inactive_user_sweep_interval_secs: u64,
     /// Byte budget for the compiled-WASM **contract** module cache. The
     /// delegate cache gets a fraction of this
     /// (`DELEGATE_MODULE_CACHE_BUDGET_DIVISOR`), so the combined ceiling is
@@ -1124,6 +1189,21 @@ fn default_per_user_secret_quota_bytes() -> u64 {
     crate::wasm_runtime::DEFAULT_PER_USER_SECRET_QUOTA_BYTES as u64
 }
 
+/// Default inactive-user TTL (30 days). Resolves to
+/// [`crate::wasm_runtime::DEFAULT_PER_USER_INACTIVE_TTL_SECS`], the single
+/// source of truth, so the operator-facing default and the sweep's fallback
+/// never drift.
+const fn default_per_user_inactive_ttl_secs() -> u64 {
+    crate::wasm_runtime::DEFAULT_PER_USER_INACTIVE_TTL_SECS
+}
+
+/// Default inactive-user sweep interval (1 hour). Far finer than the 30-day
+/// TTL, so reclamation lag is negligible while the periodic disk walk stays
+/// cheap.
+const fn default_inactive_user_sweep_interval_secs() -> u64 {
+    3_600
+}
+
 /// Default contract-module cache byte budget, scaled to system RAM
 /// (`clamp(total_ram / 8, 64 MiB, 1.5 GiB)`).
 ///
@@ -2022,6 +2102,21 @@ pub struct WebsocketApiConfig {
         rename = "per-user-export-min-interval-secs"
     )]
     pub per_user_export_min_interval_secs: u64,
+
+    /// Resolved secrets directory for this node. RUNTIME-ONLY, NOT persisted
+    /// (`#[serde(skip)]`, like `TelemetryConfig::is_test_environment`): it is
+    /// derived from the full `Config` in `build()` (`config.secrets_dir()`), so
+    /// serializing/round-tripping a `WebsocketApiConfig` standalone leaves it
+    /// empty and `build()` repopulates it.
+    ///
+    /// The WS serve layer injects it as an `Extension` so the per-user
+    /// last-activity marker (#4561, P5 of #4381, inactive-user TTL) can be
+    /// stamped at the same `<base>/users/<user_id>/.last_seen` location the
+    /// reclaim sweep reads. Empty (the default on the standalone test paths)
+    /// disables stamping, which is correct for non-hosted/test composition that
+    /// has no secrets tree to mark.
+    #[serde(skip)]
+    pub secrets_dir: std::path::PathBuf,
 }
 
 #[inline]
@@ -2067,6 +2162,7 @@ impl From<SocketAddr> for WebsocketApiConfig {
             per_user_op_rate_limit: default_per_user_op_rate_limit(),
             per_user_op_burst: default_per_user_op_burst(),
             per_user_export_min_interval_secs: default_per_user_export_min_interval_secs(),
+            secrets_dir: std::path::PathBuf::new(),
         }
     }
 }
@@ -2085,6 +2181,7 @@ impl Default for WebsocketApiConfig {
             per_user_op_rate_limit: default_per_user_op_rate_limit(),
             per_user_op_burst: default_per_user_op_burst(),
             per_user_export_min_interval_secs: default_per_user_export_min_interval_secs(),
+            secrets_dir: std::path::PathBuf::new(),
         }
     }
 }
@@ -3986,6 +4083,8 @@ mod tests {
             max_blocking_threads: None,
             max_hosting_storage: None,
             per_user_secret_quota_bytes: None,
+            per_user_inactive_ttl_secs: None,
+            inactive_user_sweep_interval_secs: None,
             module_cache_budget_bytes: None,
             shutdown_drain_secs: None,
             telemetry: Default::default(),
@@ -4044,6 +4143,9 @@ mod tests {
                 per_user_op_rate_limit: 33,
                 per_user_op_burst: 77,
                 per_user_export_min_interval_secs: 17,
+                // serde-skip runtime field; repopulated by build() and not
+                // asserted in the round-trip (bound to `_` in the destructure).
+                secrets_dir: std::path::PathBuf::new(),
             },
             secrets: base.secrets.clone(),
             log_level: tracing::log::LevelFilter::Debug,
@@ -4055,6 +4157,8 @@ mod tests {
             max_blocking_threads: 7,
             max_hosting_storage: 123_456_789,
             per_user_secret_quota_bytes: 7_654_321,
+            per_user_inactive_ttl_secs: 1_234_567,
+            inactive_user_sweep_interval_secs: 7_200,
             module_cache_budget_bytes: 987_654_321,
             telemetry: TelemetryConfig {
                 enabled: false,
@@ -4090,6 +4194,8 @@ mod tests {
             max_blocking_threads,
             max_hosting_storage,
             per_user_secret_quota_bytes,
+            per_user_inactive_ttl_secs,
+            inactive_user_sweep_interval_secs,
             module_cache_budget_bytes,
             telemetry,
             shutdown_drain_secs,
@@ -4111,6 +4217,14 @@ mod tests {
             per_user_secret_quota_bytes, seed.per_user_secret_quota_bytes,
             "per_user_secret_quota_bytes"
         );
+        assert_eq!(
+            per_user_inactive_ttl_secs, seed.per_user_inactive_ttl_secs,
+            "per_user_inactive_ttl_secs"
+        );
+        assert_eq!(
+            inactive_user_sweep_interval_secs, seed.inactive_user_sweep_interval_secs,
+            "inactive_user_sweep_interval_secs"
+        );
         assert_eq!(
             module_cache_budget_bytes, seed.module_cache_budget_bytes,
             "module_cache_budget_bytes"
@@ -4216,6 +4330,7 @@ mod tests {
             per_user_op_rate_limit,
             per_user_op_burst,
             per_user_export_min_interval_secs,
+            secrets_dir: _, // serde-skip runtime field, repopulated by build()
         } = ws_api;
         assert_eq!(ws_address, seed.ws_api.address, "ws_api.address");
         assert_eq!(ws_port, seed.ws_api.port, "ws_api.port");