Refactor ETag generation for oneliners API

Remove the per-user oneliner watermark, which was previously used to
generate ETags for the `/api/oneliners` endpoint. The watermark was
intended to track the last successful oneliner write for each user.

The ETag generation has been refactored to use a deterministic FNV-1a
hash. This hash is computed over a sorted list of tuples containing
`(case_id, last_recording_at_ns, oneliner_mtime_ns)` for all visible
cases. This approach ensures that the ETag changes whenever any
listen-relevant file system modification occurs, such as a new
recording, oneliner regeneration, soft-delete, undo, or
upload-auto-reopen. This new method is more robust and avoids the
complexity of managing per-user state.

The `OnelinerWatermark` type alias and its associated logic have been
removed from `AppState` and the relevant modules (`transcribe::worker`,
`transcribe::recovery`, `routes::oneliners`, `main`). New integration
tests have been added to verify that various delete operations (single
delete, undo delete, bulk delete) correctly trigger an ETag update,
ensuring cache invalidation.
This commit is contained in:
2026-04-19 14:47:30 +02:00
parent 5e8d2f8e58
commit a1ff410d1b
11 changed files with 425 additions and 80 deletions
+6 -2
View File
@@ -41,8 +41,12 @@ pub struct CaseMarker {
/// updates. Drives the UI sort order (newest activity first). /// updates. Drives the UI sort order (newest activity first).
pub last_activity_at: String, pub last_activity_at: String,
/// `true` once this case has been seen in any `/api/oneliners` /// `true` once this case has been seen in any `/api/oneliners`
/// response. Reserved for future delete-propagation logic; today only /// response or after a successful upload ACK. Drives two clean-up
/// useful for diagnostics. /// paths: [`CaseStore::reconcile_with_server_snapshot`] only removes
/// synced markers the server no longer lists, and
/// [`CaseStore::cleanup_stale`] only ages out synced markers. Pending
/// (`= false`) markers guard unsent uploads and are never touched by
/// either — losing one would silently erase a patient recording.
pub synced_to_server: bool, pub synced_to_server: bool,
/// `None` while transcription / oneliner generation is pending; set /// `None` while transcription / oneliner generation is pending; set
/// from the server snapshot when available. /// from the server snapshot when available.
-18
View File
@@ -9,27 +9,17 @@ pub mod routes;
pub mod transcribe; pub mod transcribe;
pub mod web_session; pub mod web_session;
use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc; use std::sync::Arc;
use axum::extract::FromRef; use axum::extract::FromRef;
use axum::Router; use axum::Router;
use time::OffsetDateTime;
use tokio::sync::RwLock;
use analyze::AnalyzeSender; use analyze::AnalyzeSender;
use config::Config; use config::Config;
use transcribe::TranscribeSender; use transcribe::TranscribeSender;
use web_session::SessionStore; use web_session::SessionStore;
/// Per-user UTC timestamp of the last successful oneliner write.
/// Read on every `/api/oneliners` request to build the ETag; written
/// from the transcribe worker and the boot recovery scan. Missing entry
/// → the user has no known oneliner activity yet; the endpoint returns
/// a full response and sets an ETag based on a `0` seed.
pub type OnelinerWatermark = Arc<RwLock<HashMap<String, OffsetDateTime>>>;
/// Live "is this worker currently processing a job?" flag. /// Live "is this worker currently processing a job?" flag.
/// Set true before each job, false after. UI uses it to distinguish a real /// Set true before each job, false after. UI uses it to distinguish a real
/// in-flight job from a stale on-disk marker (orphan input, missing /// in-flight job from a stale on-disk marker (orphan input, missing
@@ -70,7 +60,6 @@ pub struct AppState {
pub session_store: SessionStore, pub session_store: SessionStore,
pub analyze_busy: AnalyzeBusy, pub analyze_busy: AnalyzeBusy,
pub transcribe_busy: TranscribeBusy, pub transcribe_busy: TranscribeBusy,
pub oneliner_watermark: OnelinerWatermark,
} }
impl FromRef<AppState> for Arc<Config> { impl FromRef<AppState> for Arc<Config> {
@@ -109,12 +98,6 @@ impl FromRef<AppState> for TranscribeBusy {
} }
} }
impl FromRef<AppState> for OnelinerWatermark {
fn from_ref(state: &AppState) -> Self {
state.oneliner_watermark.clone()
}
}
/// Test/simple entrypoint: jobs pushed into either channel are dropped /// Test/simple entrypoint: jobs pushed into either channel are dropped
/// because the receivers are not retained. Use [`create_router_with_state`] /// because the receivers are not retained. Use [`create_router_with_state`]
/// from `main.rs` where real workers own the receivers. /// from `main.rs` where real workers own the receivers.
@@ -128,7 +111,6 @@ pub fn create_router(config: Arc<Config>) -> Router {
session_store: web_session::new_store(), session_store: web_session::new_store(),
analyze_busy: AnalyzeBusy(Arc::new(AtomicBool::new(false))), analyze_busy: AnalyzeBusy(Arc::new(AtomicBool::new(false))),
transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))), transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))),
oneliner_watermark: Arc::new(RwLock::new(HashMap::new())),
}) })
} }
+2 -12
View File
@@ -12,7 +12,7 @@ use doctate_server::analyze;
use doctate_server::config::Config; use doctate_server::config::Config;
use doctate_server::gazetteer::{Gazetteer, SpellbookDict}; use doctate_server::gazetteer::{Gazetteer, SpellbookDict};
use doctate_server::transcribe; use doctate_server::transcribe;
use doctate_server::{AppState, OnelinerWatermark}; use doctate_server::AppState;
#[tokio::main] #[tokio::main]
async fn main() { async fn main() {
@@ -110,13 +110,6 @@ async fn main() {
} }
}); });
// Oneliner watermark: per-user "last successful oneliner write (UTC)".
// Feeds the ETag on `/api/oneliners`. Shared across the transcribe
// worker (writer) and the oneliners route handler (reader).
let oneliner_watermark: OnelinerWatermark = Arc::new(tokio::sync::RwLock::new(
std::collections::HashMap::new(),
));
// Transcription pipeline: channel + worker + recovery scan. // Transcription pipeline: channel + worker + recovery scan.
let (transcribe_tx, transcribe_rx) = transcribe::channel(); let (transcribe_tx, transcribe_rx) = transcribe::channel();
let transcribe_busy: doctate_server::WorkerBusy = let transcribe_busy: doctate_server::WorkerBusy =
@@ -127,7 +120,6 @@ async fn main() {
http_client.clone(), http_client.clone(),
transcribe_busy.clone(), transcribe_busy.clone(),
vocab.clone(), vocab.clone(),
oneliner_watermark.clone(),
)); ));
{ {
let tx = transcribe_tx.clone(); let tx = transcribe_tx.clone();
@@ -135,13 +127,12 @@ async fn main() {
let client = http_client.clone(); let client = http_client.clone();
let cfg = config.clone(); let cfg = config.clone();
let vocab = vocab.clone(); let vocab = vocab.clone();
let watermark = oneliner_watermark.clone();
tokio::spawn(async move { tokio::spawn(async move {
transcribe::recovery::scan_and_enqueue(&data_path, &tx).await; transcribe::recovery::scan_and_enqueue(&data_path, &tx).await;
// Then catch up oneliners for cases that crashed between // Then catch up oneliners for cases that crashed between
// transcript-write and oneliner-write in a previous run. // transcript-write and oneliner-write in a previous run.
transcribe::recovery::regenerate_missing_oneliners( transcribe::recovery::regenerate_missing_oneliners(
&data_path, &client, &cfg, &vocab, &watermark, &data_path, &client, &cfg, &vocab,
) )
.await; .await;
}); });
@@ -173,7 +164,6 @@ async fn main() {
session_store: doctate_server::web_session::new_store(), session_store: doctate_server::web_session::new_store(),
analyze_busy: doctate_server::AnalyzeBusy(analyze_busy), analyze_busy: doctate_server::AnalyzeBusy(analyze_busy),
transcribe_busy: doctate_server::TranscribeBusy(transcribe_busy), transcribe_busy: doctate_server::TranscribeBusy(transcribe_busy),
oneliner_watermark,
}; };
let addr = format!("0.0.0.0:{}", config.server_port); let addr = format!("0.0.0.0:{}", config.server_port);
+96 -19
View File
@@ -5,16 +5,22 @@
//! text (or null if not yet generated), and UTC timestamps for when the //! text (or null if not yet generated), and UTC timestamps for when the
//! case started and when its oneliner was last written. //! case started and when its oneliner was last written.
//! //!
//! Conditional-GET: the response carries a weak ETag `W/"<unix>-<hours>"` //! Conditional-GET: the response carries a weak ETag
//! derived from the per-user watermark. Clients that send back a matching //! `W/"<fingerprint_hex>-<hours>"`. The fingerprint is a deterministic
//! `If-None-Match` get `304 Not Modified` without touching the filesystem. //! FNV-1a hash over the sorted tuples
//! `(case_id, last_recording_at_ns, oneliner_mtime_ns)` of every
//! *visible* case. Any listen-relevant FS change — new recording,
//! oneliner regeneration, soft-delete, undo, upload-auto-reopen —
//! shifts at least one tuple and the hash moves. Server restart is a
//! no-op: same FS, same hash.
//!
//! The `hours` component is included so a client swapping window sizes //! The `hours` component is included so a client swapping window sizes
//! does not get a stale 304. //! does not get a stale 304.
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::time::SystemTime; use std::time::SystemTime;
use axum::extract::{Query, State}; use axum::extract::Query;
use axum::http::{header, HeaderMap, StatusCode}; use axum::http::{header, HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response}; use axum::response::{IntoResponse, Response};
use axum::Json; use axum::Json;
@@ -26,7 +32,6 @@ use tracing::warn;
use crate::auth::AuthenticatedUser; use crate::auth::AuthenticatedUser;
use crate::error::AppError; use crate::error::AppError;
use crate::OnelinerWatermark;
const DEFAULT_HOURS: u32 = 16; const DEFAULT_HOURS: u32 = 16;
const MAX_HOURS: u32 = 168; const MAX_HOURS: u32 = 168;
@@ -38,7 +43,6 @@ pub struct OnelinersQuery {
pub async fn handle_oneliners( pub async fn handle_oneliners(
user: AuthenticatedUser, user: AuthenticatedUser,
State(watermark): State<OnelinerWatermark>,
Query(params): Query<OnelinersQuery>, Query(params): Query<OnelinersQuery>,
headers: HeaderMap, headers: HeaderMap,
) -> Result<Response, AppError> { ) -> Result<Response, AppError> {
@@ -46,18 +50,21 @@ pub async fn handle_oneliners(
let now = OffsetDateTime::now_utc(); let now = OffsetDateTime::now_utc();
let since = now - Duration::hours(hours as i64); let since = now - Duration::hours(hours as i64);
let current_watermark = watermark.read().await.get(&user.slug).copied(); // One scan pass yields both the case list and the fingerprint —
let etag = build_etag(current_watermark, hours); // the fingerprint is the authoritative signal for "has the view
// changed?". Body serialization is skipped if the client already
// has this version (304).
let scan = scan_cases(&user.data_dir, since).await;
let etag = build_etag(scan.fingerprint, hours);
if client_has_matching_etag(&headers, &etag) { if client_has_matching_etag(&headers, &etag) {
return Ok(not_modified(&etag)); return Ok(not_modified(&etag));
} }
let oneliners = enumerate_recent_oneliners(&user.data_dir, since).await;
let body = OnelinersResponse { let body = OnelinersResponse {
as_of: now.format(&Rfc3339).unwrap_or_default(), as_of: now.format(&Rfc3339).unwrap_or_default(),
window_hours: hours, window_hours: hours,
oneliners, oneliners: scan.entries,
}; };
let mut response = Json(body).into_response(); let mut response = Json(body).into_response();
@@ -67,9 +74,8 @@ pub async fn handle_oneliners(
Ok(response) Ok(response)
} }
fn build_etag(watermark: Option<OffsetDateTime>, hours: u32) -> String { fn build_etag(fingerprint: u64, hours: u32) -> String {
let seconds = watermark.map(|t| t.unix_timestamp()).unwrap_or(0); format!("W/\"{fingerprint:016x}-{hours}\"")
format!("W/\"{seconds}-{hours}\"")
} }
/// Case-sensitive match against the raw `If-None-Match` header value. /// Case-sensitive match against the raw `If-None-Match` header value.
@@ -91,19 +97,36 @@ fn not_modified(etag: &str) -> Response {
response response
} }
/// Output of [`scan_cases`]: the sorted entry list that becomes the
/// response body, and a fingerprint derived from the same scan.
struct ScanResult {
entries: Vec<OnelinerEntry>,
fingerprint: u64,
}
/// Scan `<user_data_dir>/<case_id>/` for cases whose latest recording /// Scan `<user_data_dir>/<case_id>/` for cases whose latest recording
/// sits inside the `[since, now]` window. Returns entries sorted by /// sits inside the `[since, now]` window. Returns entries sorted by
/// `last_recording_at` descending (most recently active first); cases /// `last_recording_at` descending (most recently active first); cases
/// without recordings fall back to `created_at`. Soft-deleted cases and /// without recordings fall back to `created_at`. Soft-deleted cases and
/// non-UUID directories are filtered out. /// non-UUID directories are filtered out.
async fn enumerate_recent_oneliners( ///
user_data_dir: &Path, /// Alongside the entries we compute a deterministic fingerprint
since: OffsetDateTime, /// (`fnv1a_64`) over the sorted set of
) -> Vec<OnelinerEntry> { /// `(case_id, last_recording_at_ns, oneliner_mtime_ns)` tuples — cheap
/// to compute during the scan, and it changes exactly when the visible
/// list changes. No RAM-local state, so server restart is a no-op.
async fn scan_cases(user_data_dir: &Path, since: OffsetDateTime) -> ScanResult {
let mut entries: Vec<OnelinerEntry> = Vec::new(); let mut entries: Vec<OnelinerEntry> = Vec::new();
let Ok(mut cases) = tokio::fs::read_dir(user_data_dir).await else { let Ok(mut cases) = tokio::fs::read_dir(user_data_dir).await else {
return entries; return ScanResult {
entries,
fingerprint: fnv1a_64(b""),
};
}; };
// Parallel array keyed 1:1 with `entries` carrying the raw nanosecond
// timestamps needed for the fingerprint. Kept separate so the
// response DTO stays flat.
let mut fp_tuples: Vec<(String, i128, i128)> = Vec::new();
while let Ok(Some(case_entry)) = cases.next_entry().await { while let Ok(Some(case_entry)) = cases.next_entry().await {
let case_dir = case_entry.path(); let case_dir = case_entry.path();
if !case_dir.is_dir() { if !case_dir.is_dir() {
@@ -139,6 +162,10 @@ async fn enumerate_recent_oneliners(
let last_recording_at_str = last_recording_at.format(&Rfc3339).ok(); let last_recording_at_str = last_recording_at.format(&Rfc3339).ok();
let updated_at_str = updated_at.and_then(|t| t.format(&Rfc3339).ok()); let updated_at_str = updated_at.and_then(|t| t.format(&Rfc3339).ok());
let last_rec_ns = last_recording_at.unix_timestamp_nanos();
let oneliner_ns = updated_at.map(|t| t.unix_timestamp_nanos()).unwrap_or(0);
fp_tuples.push((case_id.to_owned(), last_rec_ns, oneliner_ns));
entries.push(OnelinerEntry { entries.push(OnelinerEntry {
case_id: case_id.to_owned(), case_id: case_id.to_owned(),
oneliner: oneliner_text, oneliner: oneliner_text,
@@ -156,9 +183,59 @@ async fn enumerate_recent_oneliners(
let b_key = b.last_recording_at.as_deref().unwrap_or(&b.created_at); let b_key = b.last_recording_at.as_deref().unwrap_or(&b.created_at);
b_key.cmp(a_key) b_key.cmp(a_key)
}); });
entries
// Fingerprint: independent of entries' display order, so sort the
// tuple list by case_id and hash.
fp_tuples.sort_by(|a, b| a.0.cmp(&b.0));
let fingerprint = fingerprint_tuples(&fp_tuples);
ScanResult {
entries,
fingerprint,
}
} }
/// FNV-1a 64-bit over the concatenation of the tuples. Stable across
/// server restarts — no `RandomState`, no HashMap hashing. Ample
/// resistance against accidental collisions for ETag purposes.
fn fingerprint_tuples(tuples: &[(String, i128, i128)]) -> u64 {
let mut h: u64 = FNV_OFFSET;
for (case_id, last_rec_ns, oneliner_ns) in tuples {
for b in case_id.as_bytes() {
h ^= *b as u64;
h = h.wrapping_mul(FNV_PRIME);
}
// Separator so `"ab" || "12"` ≠ `"a" || "b12"`.
h ^= b'|' as u64;
h = h.wrapping_mul(FNV_PRIME);
for b in last_rec_ns.to_le_bytes() {
h ^= b as u64;
h = h.wrapping_mul(FNV_PRIME);
}
h ^= b'|' as u64;
h = h.wrapping_mul(FNV_PRIME);
for b in oneliner_ns.to_le_bytes() {
h ^= b as u64;
h = h.wrapping_mul(FNV_PRIME);
}
h ^= b'\n' as u64;
h = h.wrapping_mul(FNV_PRIME);
}
h
}
fn fnv1a_64(bytes: &[u8]) -> u64 {
let mut h: u64 = FNV_OFFSET;
for b in bytes {
h ^= *b as u64;
h = h.wrapping_mul(FNV_PRIME);
}
h
}
const FNV_OFFSET: u64 = 0xcbf29ce4_84222325;
const FNV_PRIME: u64 = 0x00000100_000001b3;
/// Return `(earliest, latest)` `.m4a` mtime pair in `case_dir`. /// Return `(earliest, latest)` `.m4a` mtime pair in `case_dir`.
/// `.m4a.failed` entries count — a failed recording still marks activity. /// `.m4a.failed` entries count — a failed recording still marks activity.
/// `None` on empty directory or read error. A single recording yields /// `None` on empty directory or read error. A single recording yields
+3 -1
View File
@@ -75,7 +75,9 @@ pub async fn handle_upload(
// Auto-reopen soft-deleted cases: a new upload clears the `.deleted` // Auto-reopen soft-deleted cases: a new upload clears the `.deleted`
// marker so the case reappears in /api/oneliners and the web UI. // marker so the case reappears in /api/oneliners and the web UI.
// Hard-deleted cases (directory already gone) are handled transparently // Hard-deleted cases (directory already gone) are handled transparently
// by `resolve_case_dir` above — nothing to clear. // by `resolve_case_dir` above — nothing to clear. No explicit ETag
// bump is needed: the subsequent m4a write changes `last_recording_at`,
// which flows into the `/api/oneliners` fingerprint automatically.
if crate::paths::is_deleted(&case_dir).await { if crate::paths::is_deleted(&case_dir).await {
tokio::fs::remove_file(case_dir.join(crate::paths::DELETE_MARKER)).await?; tokio::fs::remove_file(case_dir.join(crate::paths::DELETE_MARKER)).await?;
info!( info!(
+2 -6
View File
@@ -5,8 +5,6 @@ use tracing::{info, warn};
use super::{TranscribeJob, TranscribeSender}; use super::{TranscribeJob, TranscribeSender};
use crate::config::Config; use crate::config::Config;
use crate::gazetteer::Gazetteer; use crate::gazetteer::Gazetteer;
use crate::OnelinerWatermark;
/// Walk `data_path/*/` and enqueue every recording pending transcription for /// Walk `data_path/*/` and enqueue every recording pending transcription for
/// every user. Intended for startup; the same primitive backs the per-user /// every user. Intended for startup; the same primitive backs the per-user
/// self-heal triggered by web handlers. /// self-heal triggered by web handlers.
@@ -87,8 +85,7 @@ pub async fn enqueue_pending_for_user(
/// one non-empty `*.transcript.txt` but no `oneliner.txt`, paired with the /// one non-empty `*.transcript.txt` but no `oneliner.txt`, paired with the
/// user-slug (parent directory name). Pure filesystem scan, no LLM calls — /// user-slug (parent directory name). Pure filesystem scan, no LLM calls —
/// separated from the regeneration wrapper so it can be unit-tested without /// separated from the regeneration wrapper so it can be unit-tested without
/// mocking Ollama. The slug is carried through so the caller can bump the /// mocking Ollama.
/// per-user watermark when regeneration succeeds.
pub(crate) async fn cases_needing_oneliner(data_path: &Path) -> Vec<(PathBuf, String)> { pub(crate) async fn cases_needing_oneliner(data_path: &Path) -> Vec<(PathBuf, String)> {
let mut out: Vec<(PathBuf, String)> = Vec::new(); let mut out: Vec<(PathBuf, String)> = Vec::new();
let Ok(mut users) = tokio::fs::read_dir(data_path).await else { let Ok(mut users) = tokio::fs::read_dir(data_path).await else {
@@ -152,7 +149,6 @@ pub async fn regenerate_missing_oneliners(
client: &reqwest::Client, client: &reqwest::Client,
config: &Config, config: &Config,
vocab: &Gazetteer, vocab: &Gazetteer,
watermark: &OnelinerWatermark,
) { ) {
let cases = cases_needing_oneliner(data_path).await; let cases = cases_needing_oneliner(data_path).await;
if cases.is_empty() { if cases.is_empty() {
@@ -160,7 +156,7 @@ pub async fn regenerate_missing_oneliners(
} }
info!(count = cases.len(), "Regenerating missing oneliners"); info!(count = cases.len(), "Regenerating missing oneliners");
for (case_dir, slug) in cases { for (case_dir, slug) in cases {
super::worker::update_oneliner(&case_dir, &slug, client, config, vocab, watermark).await; super::worker::update_oneliner(&case_dir, &slug, client, config, vocab).await;
} }
} }
+8 -12
View File
@@ -2,13 +2,12 @@ use std::path::Path;
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use time::OffsetDateTime;
use tracing::{error, info, warn}; use tracing::{error, info, warn};
use super::{ffmpeg, ollama, whisper, TranscribeReceiver}; use super::{ffmpeg, ollama, whisper, TranscribeReceiver};
use crate::config::{Config, WhisperUserSettings}; use crate::config::{Config, WhisperUserSettings};
use crate::gazetteer::Gazetteer; use crate::gazetteer::Gazetteer;
use crate::{BusyGuard, OnelinerWatermark, WorkerBusy}; use crate::{BusyGuard, WorkerBusy};
const ONELINER_TIMEOUT: Duration = Duration::from_secs(60); const ONELINER_TIMEOUT: Duration = Duration::from_secs(60);
@@ -20,7 +19,6 @@ pub async fn run(
client: reqwest::Client, client: reqwest::Client,
worker_busy: WorkerBusy, worker_busy: WorkerBusy,
vocab: Arc<Gazetteer>, vocab: Arc<Gazetteer>,
watermark: OnelinerWatermark,
) { ) {
info!("Transcription worker started"); info!("Transcription worker started");
let timeout = Duration::from_secs(config.whisper_timeout_seconds); let timeout = Duration::from_secs(config.whisper_timeout_seconds);
@@ -90,7 +88,7 @@ pub async fn run(
if let Some(case_dir) = audio_path.parent() if let Some(case_dir) = audio_path.parent()
&& !has_pending_recordings(case_dir).await && !has_pending_recordings(case_dir).await
{ {
update_oneliner(case_dir, &job.user_slug, &client, &config, &vocab, &watermark).await; update_oneliner(case_dir, &job.user_slug, &client, &config, &vocab).await;
} }
} }
@@ -133,7 +131,6 @@ pub(crate) async fn update_oneliner(
client: &reqwest::Client, client: &reqwest::Client,
config: &Config, config: &Config,
vocab: &Gazetteer, vocab: &Gazetteer,
watermark: &OnelinerWatermark,
) { ) {
let path = case_dir.join("oneliner.txt"); let path = case_dir.join("oneliner.txt");
@@ -157,13 +154,12 @@ pub(crate) async fn update_oneliner(
if let Err(e) = tokio::fs::write(&path, &line).await { if let Err(e) = tokio::fs::write(&path, &line).await {
error!(path = %path.display(), error = %e, "writing oneliner failed"); error!(path = %path.display(), error = %e, "writing oneliner failed");
} else { } else {
info!(path = %path.display(), chars = line.chars().count(), "Oneliner updated"); info!(
// Bump the per-user watermark so the next `/api/oneliners` user = %user_slug,
// poll sees a fresh ETag and fetches the new content. path = %path.display(),
watermark chars = line.chars().count(),
.write() "Oneliner updated"
.await );
.insert(user_slug.to_string(), OffsetDateTime::now_utc());
} }
} }
Err(e) => { Err(e) => {
+250
View File
@@ -0,0 +1,250 @@
//! Listen-relevante Mutationen (Einzel-Delete, Undo, Bulk-Delete) müssen
//! den pro-User Watermark bumpen. Sonst bleibt der ETag von
//! `/api/oneliners` stehen, jeder Poll liefert 304, und der Client
//! läuft dauerhaft mit einem Marker für einen Fall, den der Server
//! bereits gelöscht hat (ursprünglicher Bugreport: "Polymyalgia
//! rheumatica" blieb im Client sichtbar, obwohl im Web-UI bereits
//! gelöscht).
use std::collections::HashMap;
use std::sync::Arc;
use axum::body::Body;
use axum::http::{header, Request, StatusCode};
use doctate_common::constants::API_KEY_HEADER;
use doctate_common::oneliners::ONELINERS_PATH;
use doctate_server::config::{Config, User};
use tower::util::ServiceExt;
fn make_user(slug: &str, password_plain: &str) -> User {
User {
slug: slug.into(),
api_key: format!("key-{slug}"),
web_password: bcrypt::hash(password_plain, 4).unwrap(),
role: "doctor".into(),
whisper: Default::default(),
}
}
fn test_config_with_users(users: Vec<User>) -> Arc<Config> {
let data_path = std::env::temp_dir().join(format!(
"doctate-delete-wm-test-{}-{}",
std::process::id(),
uuid::Uuid::new_v4()
));
let api_keys: HashMap<String, String> =
users.iter().map(|u| (u.api_key.clone(), u.slug.clone())).collect();
Arc::new(Config {
data_path,
users,
api_keys,
..Config::test_default()
})
}
fn extract_session_cookie(resp: &axum::response::Response) -> Option<String> {
for v in resp.headers().get_all(header::SET_COOKIE).iter() {
let s = v.to_str().ok()?;
if let Some(pair) = s.split(';').next()
&& pair.starts_with("session=")
{
return Some(pair.to_string());
}
}
None
}
fn login_request(slug: &str, password: &str) -> Request<Body> {
let body = format!("slug={slug}&password={password}");
Request::builder()
.method("POST")
.uri("/web/login")
.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
.body(Body::from(body))
.unwrap()
}
fn oneliners_request(api_key: &str, if_none_match: Option<&str>) -> Request<Body> {
let mut b = Request::builder()
.method("GET")
.uri(ONELINERS_PATH)
.header(API_KEY_HEADER, api_key);
if let Some(tag) = if_none_match {
b = b.header(header::IF_NONE_MATCH, tag);
}
b.body(Body::empty()).unwrap()
}
/// Create `<data_path>/<slug>/<case_id>/<ts>.m4a` so the case is
/// visible to `enumerate_recent_oneliners` (window defaults to 16h;
/// the fake mtime is "now" because write happens right now).
fn seed_case(data_path: &std::path::Path, slug: &str, case_id: &str) {
let case_dir = data_path.join(slug).join(case_id);
std::fs::create_dir_all(&case_dir).unwrap();
std::fs::write(case_dir.join("2026-04-19T10-00-00Z.m4a"), b"fake audio").unwrap();
}
async fn capture_etag(
app: &axum::Router,
api_key: &str,
if_none_match: Option<&str>,
) -> (StatusCode, Option<String>) {
let resp = app
.clone()
.oneshot(oneliners_request(api_key, if_none_match))
.await
.unwrap();
let status = resp.status();
let etag = resp
.headers()
.get(header::ETAG)
.and_then(|v| v.to_str().ok())
.map(str::to_owned);
(status, etag)
}
#[tokio::test]
async fn delete_bumps_watermark() {
let config = test_config_with_users(vec![make_user("dr_a", "s")]);
let data_path = config.data_path.clone();
let case_id = "550e8400-e29b-41d4-a716-446655440000";
seed_case(&data_path, "dr_a", case_id);
let app = doctate_server::create_router(config);
let login = app.clone().oneshot(login_request("dr_a", "s")).await.unwrap();
let cookie = extract_session_cookie(&login).expect("login should set cookie");
// First GET: capture baseline ETag (watermark empty → seeded on 0).
let (s1, etag_before) = capture_etag(&app, "key-dr_a", None).await;
assert_eq!(s1, StatusCode::OK);
let etag_before = etag_before.expect("server must set ETag");
// Soft-delete via the web handler.
let del = app
.clone()
.oneshot(
Request::builder()
.method("POST")
.uri(format!("/web/cases/{case_id}/delete"))
.header(header::COOKIE, &cookie)
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(
del.status().as_u16() / 100,
3,
"delete should redirect (3xx)"
);
// Same If-None-Match now must NOT match — watermark was bumped.
let (s2, etag_after) = capture_etag(&app, "key-dr_a", Some(&etag_before)).await;
assert_eq!(
s2,
StatusCode::OK,
"watermark bump must invalidate cached ETag",
);
let etag_after = etag_after.expect("200 response must set ETag");
assert_ne!(
etag_after, etag_before,
"delete handler must change the per-user ETag"
);
}
#[tokio::test]
async fn undo_delete_bumps_watermark() {
let config = test_config_with_users(vec![make_user("dr_a", "s")]);
let data_path = config.data_path.clone();
let case_id = "660e8400-e29b-41d4-a716-446655440000";
seed_case(&data_path, "dr_a", case_id);
let app = doctate_server::create_router(config);
let login = app.clone().oneshot(login_request("dr_a", "s")).await.unwrap();
let cookie = extract_session_cookie(&login).expect("login should set cookie");
// Delete first so undo has something to restore; grab the post-delete ETag.
app.clone()
.oneshot(
Request::builder()
.method("POST")
.uri(format!("/web/cases/{case_id}/delete"))
.header(header::COOKIE, &cookie)
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
let (_, etag_after_delete) = capture_etag(&app, "key-dr_a", None).await;
let etag_after_delete = etag_after_delete.expect("ETag must be set");
// Undo.
let undo = app
.clone()
.oneshot(
Request::builder()
.method("POST")
.uri("/web/cases/undo-delete")
.header(header::COOKIE, &cookie)
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(undo.status().as_u16() / 100, 3, "undo should redirect");
// Undo must bump again → post-delete ETag becomes stale.
let (s, etag_after_undo) = capture_etag(&app, "key-dr_a", Some(&etag_after_delete)).await;
assert_eq!(
s,
StatusCode::OK,
"undo must change the ETag so clients re-fetch"
);
let etag_after_undo = etag_after_undo.expect("200 response must set ETag");
assert_ne!(etag_after_undo, etag_after_delete);
}
#[tokio::test]
async fn bulk_delete_bumps_watermark() {
let config = test_config_with_users(vec![make_user("dr_a", "s")]);
let data_path = config.data_path.clone();
let case_a = "770e8400-e29b-41d4-a716-446655440000";
let case_b = "880e8400-e29b-41d4-a716-446655440000";
seed_case(&data_path, "dr_a", case_a);
seed_case(&data_path, "dr_a", case_b);
let app = doctate_server::create_router(config);
let login = app.clone().oneshot(login_request("dr_a", "s")).await.unwrap();
let cookie = extract_session_cookie(&login).expect("login should set cookie");
let (_, etag_before) = capture_etag(&app, "key-dr_a", None).await;
let etag_before = etag_before.expect("ETag must be set");
// Form-encoded bulk delete: action=delete&case_id=a&case_id=b.
let form = format!("action=delete&case_id={case_a}&case_id={case_b}");
let bulk = app
.clone()
.oneshot(
Request::builder()
.method("POST")
.uri("/web/cases/bulk")
.header(header::COOKIE, &cookie)
.header(
header::CONTENT_TYPE,
"application/x-www-form-urlencoded",
)
.body(Body::from(form))
.unwrap(),
)
.await
.unwrap();
assert_eq!(bulk.status().as_u16() / 100, 3, "bulk should redirect");
let (s, etag_after) = capture_etag(&app, "key-dr_a", Some(&etag_before)).await;
assert_eq!(s, StatusCode::OK, "bulk-delete must bump the ETag");
let etag_after = etag_after.expect("200 response must set ETag");
assert_ne!(etag_after, etag_before);
}
+2 -1
View File
@@ -128,7 +128,8 @@ async fn empty_user_dir_returns_200_empty_list() {
assert_eq!(response.status(), StatusCode::OK); assert_eq!(response.status(), StatusCode::OK);
let etag = header_str(&response, "etag"); let etag = header_str(&response, "etag");
assert!(etag.starts_with("W/\"0-16\""), "etag was {etag:?}"); assert!(etag.starts_with("W/\""), "etag was {etag:?}");
assert!(etag.ends_with("-16\""), "etag was {etag:?}");
let body = body_json(response).await; let body = body_json(response).await;
assert_eq!(body["window_hours"], 16); assert_eq!(body["window_hours"], 16);
+2 -8
View File
@@ -288,10 +288,7 @@ async fn worker_renames_audio_to_failed_on_whisper_error() {
let client = reqwest::Client::new(); let client = reqwest::Client::new();
let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let vocab = std::sync::Arc::new(doctate_server::gazetteer::Gazetteer::empty()); let vocab = std::sync::Arc::new(doctate_server::gazetteer::Gazetteer::empty());
let watermark: doctate_server::OnelinerWatermark = std::sync::Arc::new( transcribe::worker::run(rx, config, client, busy, vocab).await;
tokio::sync::RwLock::new(std::collections::HashMap::new()),
);
transcribe::worker::run(rx, config, client, busy, vocab, watermark).await;
// Audio must have been renamed so recovery skips it next time. // Audio must have been renamed so recovery skips it next time.
assert!(!audio.exists(), "original .m4a still present"); assert!(!audio.exists(), "original .m4a still present");
@@ -340,10 +337,7 @@ async fn transcribe_worker_normalizes_whisper_output() {
let client = reqwest::Client::new(); let client = reqwest::Client::new();
let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
let watermark: doctate_server::OnelinerWatermark = std::sync::Arc::new( transcribe::worker::run(rx, config, client, busy, vocab).await;
tokio::sync::RwLock::new(std::collections::HashMap::new()),
);
transcribe::worker::run(rx, config, client, busy, vocab, watermark).await;
let transcript_path = case_dir.join("2026-04-13T10-30-00Z.transcript.txt"); let transcript_path = case_dir.join("2026-04-13T10-30-00Z.transcript.txt");
assert!(transcript_path.exists(), "transcript missing"); assert!(transcript_path.exists(), "transcript missing");
+54 -1
View File
@@ -284,7 +284,9 @@ async fn upload_reopens_soft_deleted_case() {
// Soft-delete the case by writing the marker directly — simulates // Soft-delete the case by writing the marker directly — simulates
// the user hitting the delete button in the web UI without having // the user hitting the delete button in the web UI without having
// to drive the /web/cases/{id}/delete handler here. // to drive the /web/cases/{id}/delete handler here. Direct write
// also means the watermark is NOT bumped here; that way the next
// GET /api/oneliners pins down the pre-reopen ETag cleanly.
let case_dir = data_path.join("dr_test").join(case_id); let case_dir = data_path.join("dr_test").join(case_id);
let marker = DeleteMarker { let marker = DeleteMarker {
batch: uuid::Uuid::new_v4(), batch: uuid::Uuid::new_v4(),
@@ -293,9 +295,32 @@ async fn upload_reopens_soft_deleted_case() {
write_delete_marker(&case_dir, &marker).await.unwrap(); write_delete_marker(&case_dir, &marker).await.unwrap();
assert!(case_dir.join(DELETE_MARKER).exists()); assert!(case_dir.join(DELETE_MARKER).exists());
// Capture the ETag before the reopen.
let etag_before = {
let resp = app
.clone()
.oneshot(
Request::builder()
.method("GET")
.uri("/api/oneliners")
.header("X-API-Key", "test-key-123")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
resp.headers()
.get("etag")
.and_then(|v| v.to_str().ok())
.map(str::to_owned)
.expect("server must set ETag")
};
// Second upload — must auto-reopen. // Second upload — must auto-reopen.
let (boundary, body) = multipart_body(case_id, "2026-04-13T12:00:00Z", b"reopen recording"); let (boundary, body) = multipart_body(case_id, "2026-04-13T12:00:00Z", b"reopen recording");
let response = app let response = app
.clone()
.oneshot( .oneshot(
Request::builder() Request::builder()
.method("POST") .method("POST")
@@ -321,6 +346,34 @@ async fn upload_reopens_soft_deleted_case() {
assert!(case_dir.join("2026-04-13T10-30-00Z.m4a").exists()); assert!(case_dir.join("2026-04-13T10-30-00Z.m4a").exists());
assert!(case_dir.join("2026-04-13T12-00-00Z.m4a").exists()); assert!(case_dir.join("2026-04-13T12-00-00Z.m4a").exists());
// Auto-reopen must bump the watermark: sending the pre-reopen
// If-None-Match must return 200 (not 304) so clients who were
// offline during the reopen notice it on their next poll.
let resp = app
.oneshot(
Request::builder()
.method("GET")
.uri("/api/oneliners")
.header("X-API-Key", "test-key-123")
.header("If-None-Match", &etag_before)
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(
resp.status(),
StatusCode::OK,
"auto-reopen must invalidate the pre-reopen ETag"
);
let etag_after = resp
.headers()
.get("etag")
.and_then(|v| v.to_str().ok())
.map(str::to_owned)
.expect("server must set ETag");
assert_ne!(etag_after, etag_before);
let _ = std::fs::remove_dir_all(&data_path); let _ = std::fs::remove_dir_all(&data_path);
} }