Phase 9 S1-S3: TTL-scoped sessions with tag leases (#105)

claw-store/src/serve_v2.rs

@@ -31,6 +31,7 @@ use crate::cluster::rpc::{
 };
 use crate::cluster::transport::{NodeIdentity, QuicClient};
 use crate::config::{Config, PeerEntry, TokenEntry};
+use crate::sessions::{LeasedTag, Session, SessionStore};
 
 /// Aggregator runtime: one QuicClient, one peer list, one identity.
 ///
@@ -63,6 +64,10 @@ pub struct V2State {
     /// entries here can be admin (no `namespace`) or scoped (must
     /// match `<namespace>:*` on tag names). First match wins.
     pub token_entries: Vec<TokenEntry>,
+    /// TTL-scoped session store (Phase 9 S1-S3). Sessions own tag
+    /// leases and are reaped by a background sweeper when their
+    /// `expires_at_unix` passes without a `renew` or `commit`.
+    pub sessions: SessionStore,
 }
 
 impl V2State {
@@ -79,6 +84,13 @@ impl V2State {
             .map_err(|e| anyhow::anyhow!("loading fleet-CA identity: {e}"))?;
         let client = QuicClient::new("0.0.0.0:0".parse()?, identity)
             .map_err(|e| anyhow::anyhow!("building QUIC client: {e}"))?;
+        // Session store persists next to warm.projects_path — the
+        // aggregator already has a writable directory there. Fall
+        // back to /tmp so tests don't need config.warm set.
+        let sessions_path = std::path::PathBuf::from(&cfg.warm.projects_path)
+            .join("aggregator-sessions.json");
+        let sessions = SessionStore::load(sessions_path)
+            .map_err(|e| anyhow::anyhow!("loading session store: {e}"))?;
         Ok(Self {
             aggregator_name: cfg.node.name.clone(),
             peers: cluster.peers.clone(),
@@ -90,6 +102,7 @@ impl V2State {
                 .as_ref()
                 .map(|a| a.tokens.clone())
                 .unwrap_or_default(),
+            sessions,
         })
     }
 
@@ -495,6 +508,7 @@ impl V2State {
             default_rpc_port_offset: self.default_rpc_port_offset,
             api_token: self.api_token.clone(),
             token_entries: self.token_entries.clone(),
+            sessions: self.sessions.clone(),
         }
     }
 }
@@ -778,13 +792,275 @@ async fn v2_auth(
     next.run(request).await
 }
 
+// ── session endpoints (Phase 9 S1-S3) ───────────────────────────
+
+#[derive(Deserialize)]
+pub struct CreateSessionBody {
+    /// Absolute-from-now TTL in seconds. Callers should send the
+    /// same value on `renew` to reset the clock; there's no additive
+    /// mode. Cap enforced at 24h to protect the sweeper from a
+    /// caller pinning tags forever.
+    pub ttl_secs: u64,
+    /// Optional human note for operator/debugging visibility.
+    #[serde(default)]
+    pub note: Option<String>,
+}
+
+#[derive(Deserialize)]
+pub struct PinInSessionBody {
+    pub tag: String,
+    pub blob_id: String,
+}
+
+#[derive(Deserialize)]
+pub struct RenewBody {
+    pub ttl_secs: u64,
+}
+
+/// Max TTL a session may live before requiring renewal or commit.
+/// Prevents a caller from creating an effectively-immortal session
+/// with, say, ttl_secs = u64::MAX.
+const MAX_SESSION_TTL_SECS: u64 = 24 * 60 * 60;
+
+/// Namespace filter that pairs a caller with the sessions they may
+/// see or modify. Admin/Open see everything; namespaced callers only
+/// see their own namespace's sessions.
+fn caller_ns(caller: &AuthedCaller) -> Option<String> {
+    match caller {
+        AuthedCaller::Admin | AuthedCaller::Open => None,
+        AuthedCaller::Namespaced { namespace } => Some(namespace.clone()),
+    }
+}
+
+/// True iff `caller` may inspect / mutate `session`.
+fn caller_may_access(caller: &AuthedCaller, session: &Session) -> bool {
+    match caller {
+        AuthedCaller::Admin | AuthedCaller::Open => true,
+        AuthedCaller::Namespaced { namespace } => {
+            session.namespace.as_deref() == Some(namespace.as_str())
+        }
+    }
+}
+
+async fn handle_create_session(
+    State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
+    Json(body): Json<CreateSessionBody>,
+) -> Result<Json<Session>, (StatusCode, String)> {
+    if body.ttl_secs == 0 || body.ttl_secs > MAX_SESSION_TTL_SECS {
+        return Err((
+            StatusCode::BAD_REQUEST,
+            format!("ttl_secs must be 1..={MAX_SESSION_TTL_SECS}"),
+        ));
+    }
+    let ns = caller_ns(&caller);
+    let sess = s
+        .sessions
+        .create(ns, body.ttl_secs, body.note)
+        .await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+    Ok(Json(sess))
+}
+
+async fn handle_list_sessions(
+    State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
+) -> Json<Vec<Session>> {
+    // caller_ns == None means admin/open → no filter → see all.
+    let filter = caller_ns(&caller);
+    let list = s.sessions.list(filter.as_deref()).await;
+    Json(list)
+}
+
+async fn handle_get_session(
+    Path(id): Path<String>,
+    State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
+) -> Result<Json<Session>, StatusCode> {
+    let sess = s.sessions.get(&id).await.ok_or(StatusCode::NOT_FOUND)?;
+    if !caller_may_access(&caller, &sess) {
+        // Preserve the "does it exist" secret from other tenants —
+        // NOT_FOUND, not FORBIDDEN. Same trick the wizard-side
+        // publish-approval endpoint uses.
+        return Err(StatusCode::NOT_FOUND);
+    }
+    Ok(Json(sess))
+}
+
+/// Pin a tag *within* a session: fan the pin out to peers, then
+/// record it in the session so a later commit/cancel/expiry can
+/// reverse it. Same tag scope rule as bare `POST /tags/:name` —
+/// namespaced callers must stay in their namespace.
+async fn handle_session_pin(
+    Path(id): Path<String>,
+    State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
+    Json(body): Json<PinInSessionBody>,
+) -> Result<Json<Session>, (StatusCode, String)> {
+    check_tag_scope(&caller, &body.tag)?;
+    let value = decode_blob_id(&body.blob_id)
+        .ok_or_else(|| (StatusCode::BAD_REQUEST, "blob_id must be 64-char hex".into()))?;
+    let sess = s
+        .sessions
+        .get(&id)
+        .await
+        .ok_or((StatusCode::NOT_FOUND, "session not found".into()))?;
+    if !caller_may_access(&caller, &sess) {
+        return Err((StatusCode::NOT_FOUND, "session not found".into()));
+    }
+    if sess.committed {
+        return Err((
+            StatusCode::CONFLICT,
+            "cannot attach leases to a committed session".into(),
+        ));
+    }
+    // Fan out the pin BEFORE attaching to the session — if the fleet
+    // rejects it, we don't record a phantom lease that can never be
+    // unpinned. Same shape as the standalone put-tag path.
+    let peer_results = fanout_put_tag(&s, &body.tag, &value).await;
+    if !peer_results.iter().all(|r| r.ok) {
+        // Return the per-peer error detail so the caller can act.
+        let err = serde_json::to_string(&peer_results).unwrap_or_default();
+        return Err((StatusCode::BAD_GATEWAY, err));
+    }
+    let lease = LeasedTag {
+        tag: body.tag,
+        blob_id_hex: body.blob_id,
+        pinned_at_unix: now_unix_u64(),
+    };
+    let updated = s
+        .sessions
+        .attach_lease(&id, lease)
+        .await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+    Ok(Json(updated))
+}
+
+async fn handle_renew_session(
+    Path(id): Path<String>,
+    State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
+    Json(body): Json<RenewBody>,
+) -> Result<Json<Session>, (StatusCode, String)> {
+    if body.ttl_secs == 0 || body.ttl_secs > MAX_SESSION_TTL_SECS {
+        return Err((
+            StatusCode::BAD_REQUEST,
+            format!("ttl_secs must be 1..={MAX_SESSION_TTL_SECS}"),
+        ));
+    }
+    let sess = s
+        .sessions
+        .get(&id)
+        .await
+        .ok_or((StatusCode::NOT_FOUND, "session not found".into()))?;
+    if !caller_may_access(&caller, &sess) {
+        return Err((StatusCode::NOT_FOUND, "session not found".into()));
+    }
+    let updated = s
+        .sessions
+        .renew(&id, body.ttl_secs)
+        .await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+    Ok(Json(updated))
+}
+
+async fn handle_commit_session(
+    Path(id): Path<String>,
+    State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
+) -> Result<Json<Session>, (StatusCode, String)> {
+    let sess = s
+        .sessions
+        .get(&id)
+        .await
+        .ok_or((StatusCode::NOT_FOUND, "session not found".into()))?;
+    if !caller_may_access(&caller, &sess) {
+        return Err((StatusCode::NOT_FOUND, "session not found".into()));
+    }
+    let updated = s
+        .sessions
+        .commit(&id)
+        .await
+        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
+    Ok(Json(updated))
+}
+
+async fn handle_delete_session(
+    Path(id): Path<String>,
+    State(s): State<Arc<V2State>>,
+    axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
+) -> Result<Json<Vec<PeerResult>>, StatusCode> {
+    let sess = s.sessions.get(&id).await.ok_or(StatusCode::NOT_FOUND)?;
+    if !caller_may_access(&caller, &sess) {
+        return Err(StatusCode::NOT_FOUND);
+    }
+    // Unpin every tag on every peer FIRST — if we removed the session
+    // first, a mid-deletion crash would strand tags on the fleet.
+    let mut all_results = Vec::new();
+    for lease in &sess.leases {
+        let r = fanout_delete_tag(&s, &lease.tag).await;
+        all_results.extend(r);
+    }
+    let _ = s.sessions.remove(&id).await;
+    Ok(Json(all_results))
+}
+
+fn now_unix_u64() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .unwrap_or(0)
+}
+
+/// Reap callback for the sessions sweeper. Owns the actual fan-out.
+/// Kept a free function so `sessions::spawn_sweeper` doesn't have
+/// to know about `V2State` or its RPC client.
+async fn reap_expired(state: Arc<V2State>, sess: Session) {
+    for lease in &sess.leases {
+        let results = fanout_delete_tag(&state, &lease.tag).await;
+        // Log per-peer failures; the sweeper is best-effort. A
+        // failing peer will be retried on the next lease that
+        // touches it because the tag stays in the store until
+        // successfully removed everywhere (well, the sweeper drops
+        // the session either way — this is a known tradeoff:
+        // durability of unpin vs. never-ending sweeper retries).
+        for r in results {
+            if !r.ok {
+                eprintln!(
+                    "sweeper: session {} tag {} peer {} unpin failed: {}",
+                    sess.id, lease.tag, r.peer,
+                    r.error.as_deref().unwrap_or("<no error message>")
+                );
+            }
+        }
+    }
+}
+
 // ── route registration ──────────────────────────────────────────
 
 /// Assemble the aggregator router with state and middleware baked in.
 /// The auth middleware needs the concrete `Arc<V2State>` at layer time
 /// (so it can read `api_token`), which is why this returns a fully-
 /// stated `Router<()>` instead of a state-generic router.
+///
+/// Also spawns the background TTL sweeper (Phase 9 S1). The task is
+/// detached — its lifetime is the process lifetime.
 pub fn build(state: Arc<V2State>) -> Router {
+    // Spawn the sweeper. 15s tick is a reasonable balance: quick
+    // enough that a mid-wizard-close cleanup feels prompt, slow
+    // enough that idle aggregators aren't burning cycles.
+    {
+        let state_for_sweeper = state.clone();
+        let store = state.sessions.clone();
+        crate::sessions::spawn_sweeper(
+            store,
+            Duration::from_secs(15),
+            move |sess| {
+                let state = state_for_sweeper.clone();
+                async move { reap_expired(state, sess).await }
+            },
+        );
+    }
     Router::new()
         .route("/api/v2/fleet", get(handle_fleet))
         .route("/api/v2/node/:name/status", get(handle_node_status))
@@ -798,6 +1074,17 @@ pub fn build(state: Arc<V2State>) -> Router {
             "/api/v2/tags/:name",
             post(handle_put_tag).delete(handle_delete_tag),
         )
+        .route(
+            "/api/v2/sessions",
+            post(handle_create_session).get(handle_list_sessions),
+        )
+        .route(
+            "/api/v2/sessions/:id",
+            get(handle_get_session).delete(handle_delete_session),
+        )
+        .route("/api/v2/sessions/:id/pin", post(handle_session_pin))
+        .route("/api/v2/sessions/:id/renew", post(handle_renew_session))
+        .route("/api/v2/sessions/:id/commit", post(handle_commit_session))
         .route_layer(axum::middleware::from_fn_with_state(state.clone(), v2_auth))
         .with_state(state)
 }