Brummel/doctate
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor: Implement oneliner poller background task

Browse Source
This commit is contained in:
Brummel
2026-04-18 10:07:28 +02:00
parent 1b4864215b
commit 9b9d68402f
1 changed files with 518 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doctate-client-core/src/oneliner_poller.rs
+518 -8
View File
@@ -1,26 +1,536 @@
//! Placeholder — full implementation follows in next step.
//! Background poller for `GET /api/oneliners`.
//!
//! Hits the server every `poll_interval` with `If-None-Match`, merges
//! 200-responses into the case store, persists each fresh response into
//! the snapshot cache, and survives transient errors by logging + waiting
//! for the next tick.
//!
//! The polling logic is factored into a standalone `poll_once` method on
//! [`PollerState`] so tests can drive a single cycle without wall-clock
//! timing. The [`OnelinerPoller`] handle just owns an [`AbortHandle`]:
//! dropping the handle aborts the task.
use std::sync::Arc;
use std::time::{Duration, Instant};
use doctate_common::oneliners::{OnelinersResponse, ONELINERS_PATH};
use doctate_common::API_KEY_HEADER;
use reqwest::StatusCode;
use tokio::task::AbortHandle;
use tracing::{debug, info, warn};
use crate::case_store::CaseStore;
use crate::snapshot_cache::{CachedSnapshot, SnapshotCache};
#[derive(Debug, Clone)]
pub struct PollerConfig {
pub server_url: String,
pub api_key: String,
pub poll_interval: std::time::Duration,
pub poll_interval: Duration,
pub window_hours: u32,
}
/// Handle that aborts the background task when dropped.
pub struct OnelinerPoller {
_abort: tokio::task::AbortHandle,
abort: AbortHandle,
/// Timestamp of the last successful 200 or 304 response. UI reads
/// this via `last_success` to render a staleness indicator.
last_success: Arc<std::sync::Mutex<Option<Instant>>>,
}
impl OnelinerPoller {
/// To be implemented.
#[doc(hidden)]
pub fn from_handle(abort: tokio::task::AbortHandle) -> Self {
Self { _abort: abort }
/// Start the background polling task. Returns immediately.
///
/// Expects a Tokio runtime context (call inside `Runtime::enter()`
/// or `#[tokio::main]`).
pub fn spawn(
http: Arc<reqwest::Client>,
config: PollerConfig,
store: Arc<CaseStore>,
cache: Arc<SnapshotCache>,
) -> Self {
let last_success = Arc::new(std::sync::Mutex::new(None));
let last_success_for_task = last_success.clone();
let handle = tokio::spawn(run_loop(
http,
config,
store,
cache,
last_success_for_task,
));
Self {
abort: handle.abort_handle(),
last_success,
}
}
/// `Some(Instant)` if the poller has ever received a 200 or 304
/// response; `None` on cold start with no successful poll yet.
pub fn last_success(&self) -> Option<Instant> {
*self.last_success.lock().expect("poisoned")
}
}
impl Drop for OnelinerPoller {
fn drop(&mut self) {
self._abort.abort();
self.abort.abort();
}
}
/// Task loop: load cache, then poll every `poll_interval` until aborted.
async fn run_loop(
http: Arc<reqwest::Client>,
config: PollerConfig,
store: Arc<CaseStore>,
cache: Arc<SnapshotCache>,
last_success: Arc<std::sync::Mutex<Option<Instant>>>,
) {
info!(
interval_s = config.poll_interval.as_secs(),
window_h = config.window_hours,
"oneliner poller started"
);
let mut state = PollerState {
http,
config,
store,
cache,
etag: None,
};
// Prime the store + ETag from disk cache before the first request.
state.prime_from_cache().await;
// Fire an immediate poll so the UI doesn't wait a full interval.
tick(&mut state, &last_success).await;
let mut interval = tokio::time::interval(state.config.poll_interval);
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
// The first tick is immediate (already covered above); consume it.
interval.tick().await;
loop {
interval.tick().await;
tick(&mut state, &last_success).await;
}
}
async fn tick(
state: &mut PollerState,
last_success: &Arc<std::sync::Mutex<Option<Instant>>>,
) {
match state.poll_once().await {
Ok(outcome) => {
if matches!(outcome, PollOutcome::Success | PollOutcome::NotModified) {
*last_success.lock().expect("poisoned") = Some(Instant::now());
}
}
Err(e) => {
warn!(error = %e, "poll error — will retry on next tick");
}
}
}
#[derive(Debug, PartialEq, Eq)]
pub enum PollOutcome {
/// Server returned 200 with a fresh snapshot — merged into the store.
Success,
/// Server returned 304 Not Modified — no change.
NotModified,
/// Transient HTTP failure (non-2xx non-304); retried next tick.
TransientHttp(u16),
}
#[derive(Debug, thiserror::Error)]
pub enum PollError {
#[error("http: {0}")]
Http(#[from] reqwest::Error),
#[error("deserialize response: {0}")]
Parse(#[from] serde_json::Error),
#[error("cache write: {0}")]
Cache(std::io::Error),
#[error("case store: {0}")]
Store(#[from] crate::case_store::CaseStoreError),
#[error("format timestamp: {0}")]
Format(#[from] time::error::Format),
}
/// Driver for a single poll cycle. Holds the long-lived `reqwest::Client`,
/// the config (unchanging over the task's lifetime), the store and cache
/// (shared with the UI), and the last ETag seen from the server.
pub(crate) struct PollerState {
http: Arc<reqwest::Client>,
config: PollerConfig,
store: Arc<CaseStore>,
cache: Arc<SnapshotCache>,
etag: Option<String>,
}
impl PollerState {
/// Load a cached snapshot (if any) from disk, merge it into the store
/// and adopt its ETag. Runs once at task start so the UI sees data
/// before the first network round-trip completes.
async fn prime_from_cache(&mut self) {
let Some(cached) = self.cache.load().await else {
return;
};
if let Err(e) = self.store.merge_server_snapshot(&cached.response).await {
warn!(error = %e, "priming store from cache failed");
}
self.etag = Some(cached.etag);
debug!("store primed from cache");
}
/// Issue one request, update store/cache/etag accordingly. Returns
/// what happened so the caller can track success timestamps.
pub(crate) async fn poll_once(&mut self) -> Result<PollOutcome, PollError> {
let url = format!(
"{}{}?hours={}",
self.config.server_url.trim_end_matches('/'),
ONELINERS_PATH,
self.config.window_hours
);
let mut req = self
.http
.get(&url)
.header(API_KEY_HEADER, &self.config.api_key);
if let Some(etag) = &self.etag {
req = req.header("If-None-Match", etag);
}
let resp = req.send().await?;
let status = resp.status();
match status {
StatusCode::OK => {
let new_etag = extract_etag(&resp);
let body: OnelinersResponse = resp.json().await?;
self.store.merge_server_snapshot(&body).await?;
if let Some(etag) = new_etag.clone() {
let cached = CachedSnapshot {
etag: etag.clone(),
fetched_at: doctate_common::now_rfc3339(),
response: body,
};
if let Err(e) = self.cache.store(&cached).await {
warn!(error = %e, "snapshot cache write failed — continuing");
}
self.etag = Some(etag);
} else {
// Server without ETag: accept body but don't try
// conditional requests. Unexpected from our own
// server; defensive fallback.
warn!("server returned 200 without ETag — conditional requests disabled");
self.etag = None;
}
Ok(PollOutcome::Success)
}
StatusCode::NOT_MODIFIED => Ok(PollOutcome::NotModified),
other => Ok(PollOutcome::TransientHttp(other.as_u16())),
}
}
}
fn extract_etag(resp: &reqwest::Response) -> Option<String> {
resp.headers()
.get("etag")
.and_then(|v| v.to_str().ok())
.map(str::to_owned)
}
#[cfg(test)]
mod tests {
use super::*;
use doctate_common::oneliners::OnelinerEntry;
use tempfile::TempDir;
use wiremock::matchers::{header, header_exists, method, path, query_param};
use wiremock::{Mock, MockServer, ResponseTemplate};
const TEST_KEY: &str = "test-key-poller";
const TEST_ETAG: &str = "W/\"42-72\"";
async fn build_state(
tmp: &TempDir,
server_uri: String,
) -> (PollerState, Arc<CaseStore>, Arc<SnapshotCache>) {
let (store_inner, _rx) = CaseStore::open(tmp.path().join("cases")).await.unwrap();
let store = Arc::new(store_inner);
let cache = Arc::new(SnapshotCache::new(tmp.path().join("cache.json")));
let state = PollerState {
http: Arc::new(reqwest::Client::new()),
config: PollerConfig {
server_url: server_uri,
api_key: TEST_KEY.into(),
poll_interval: Duration::from_secs(5),
window_hours: 72,
},
store: store.clone(),
cache: cache.clone(),
etag: None,
};
(state, store, cache)
}
fn response_body(case_id: &str, oneliner: Option<&str>) -> OnelinersResponse {
OnelinersResponse {
as_of: "2026-04-18T10:00:00Z".into(),
window_hours: 72,
oneliners: vec![OnelinerEntry {
case_id: case_id.into(),
oneliner: oneliner.map(str::to_owned),
created_at: "2026-04-18T09:30:00Z".into(),
updated_at: Some("2026-04-18T09:45:00Z".into()),
}],
}
}
#[tokio::test]
async fn poll_once_on_200_merges_store_and_caches() {
let server = MockServer::start().await;
let case_id = "550e8400-e29b-41d4-a716-000000000001";
Mock::given(method("GET"))
.and(path(ONELINERS_PATH))
.and(query_param("hours", "72"))
.and(header(API_KEY_HEADER, TEST_KEY))
.respond_with(
ResponseTemplate::new(200)
.insert_header("etag", TEST_ETAG)
.set_body_json(response_body(case_id, Some("Kniegelenk"))),
)
.mount(&server)
.await;
let tmp = TempDir::new().unwrap();
let (mut state, store, cache) = build_state(&tmp, server.uri()).await;
assert_eq!(state.poll_once().await.unwrap(), PollOutcome::Success);
let list = store.list().await;
assert_eq!(list.len(), 1);
assert_eq!(list[0].oneliner.as_deref(), Some("Kniegelenk"));
assert!(list[0].synced_to_server);
let cached = cache.load().await.expect("cache written");
assert_eq!(cached.etag, TEST_ETAG);
assert_eq!(state.etag.as_deref(), Some(TEST_ETAG));
}
#[tokio::test]
async fn poll_once_sends_if_none_match_when_etag_known() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(ONELINERS_PATH))
.and(header("If-None-Match", TEST_ETAG))
.respond_with(ResponseTemplate::new(304).insert_header("etag", TEST_ETAG))
.expect(1)
.mount(&server)
.await;
let tmp = TempDir::new().unwrap();
let (mut state, _store, _cache) = build_state(&tmp, server.uri()).await;
state.etag = Some(TEST_ETAG.into());
assert_eq!(state.poll_once().await.unwrap(), PollOutcome::NotModified);
// Mock's `.expect(1)` enforces the conditional header was sent.
}
#[tokio::test]
async fn poll_once_first_call_has_no_if_none_match() {
let server = MockServer::start().await;
let case_id = "550e8400-e29b-41d4-a716-000000000002";
// Only a mock that does NOT require If-None-Match matches; if the
// poller sent one, wiremock would return 404.
Mock::given(method("GET"))
.and(path(ONELINERS_PATH))
.respond_with(
ResponseTemplate::new(200)
.insert_header("etag", TEST_ETAG)
.set_body_json(response_body(case_id, Some("x"))),
)
.expect(1)
.mount(&server)
.await;
let tmp = TempDir::new().unwrap();
let (mut state, _store, _cache) = build_state(&tmp, server.uri()).await;
assert!(state.etag.is_none());
assert_eq!(state.poll_once().await.unwrap(), PollOutcome::Success);
}
#[tokio::test]
async fn poll_once_on_304_does_not_overwrite_store() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.respond_with(ResponseTemplate::new(304).insert_header("etag", TEST_ETAG))
.mount(&server)
.await;
let tmp = TempDir::new().unwrap();
let (mut state, store, cache) = build_state(&tmp, server.uri()).await;
// Seed store with a local marker — must survive a 304.
let id = uuid::Uuid::new_v4();
store
.create_local(
id,
time::OffsetDateTime::parse(
"2026-04-18T10:00:00Z",
&time::format_description::well_known::Rfc3339,
)
.unwrap(),
)
.await
.unwrap();
assert_eq!(state.poll_once().await.unwrap(), PollOutcome::NotModified);
assert_eq!(store.list().await.len(), 1);
assert!(cache.load().await.is_none(), "304 must not create cache");
}
#[tokio::test]
async fn poll_once_returns_transient_on_500() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.respond_with(ResponseTemplate::new(500))
.mount(&server)
.await;
let tmp = TempDir::new().unwrap();
let (mut state, _store, _cache) = build_state(&tmp, server.uri()).await;
assert_eq!(
state.poll_once().await.unwrap(),
PollOutcome::TransientHttp(500)
);
}
#[tokio::test]
async fn poll_once_returns_transient_on_401() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.respond_with(ResponseTemplate::new(401))
.mount(&server)
.await;
let tmp = TempDir::new().unwrap();
let (mut state, _store, _cache) = build_state(&tmp, server.uri()).await;
// 401 bubbles up as TransientHttp so the task keeps running; the
// user surfaces the wrong-API-key state via the config panel, not
// via a fatal poller crash.
assert_eq!(
state.poll_once().await.unwrap(),
PollOutcome::TransientHttp(401)
);
}
#[tokio::test]
async fn poll_once_propagates_network_error() {
let tmp = TempDir::new().unwrap();
// Unbound address → connection refused.
let (mut state, _store, _cache) =
build_state(&tmp, "http://127.0.0.1:1".into()).await;
let err = state.poll_once().await.unwrap_err();
assert!(matches!(err, PollError::Http(_)));
}
#[tokio::test]
async fn prime_from_cache_seeds_store_and_etag() {
let tmp = TempDir::new().unwrap();
let (mut state, store, cache) = build_state(&tmp, "http://unused".into()).await;
let case_id = "550e8400-e29b-41d4-a716-000000000003";
cache
.store(&CachedSnapshot {
etag: TEST_ETAG.into(),
fetched_at: "2026-04-18T10:00:00Z".into(),
response: response_body(case_id, Some("cached")),
})
.await
.unwrap();
state.prime_from_cache().await;
assert_eq!(state.etag.as_deref(), Some(TEST_ETAG));
let list = store.list().await;
assert_eq!(list.len(), 1);
assert_eq!(list[0].oneliner.as_deref(), Some("cached"));
}
#[tokio::test]
async fn spawn_and_drop_aborts_task() {
let server = MockServer::start().await;
// Respond to any GET so the loop has work to do.
Mock::given(method("GET"))
.respond_with(
ResponseTemplate::new(200)
.insert_header("etag", TEST_ETAG)
.set_body_json(response_body(
"550e8400-e29b-41d4-a716-000000000004",
Some("x"),
)),
)
.mount(&server)
.await;
let tmp = TempDir::new().unwrap();
let (store_inner, _rx) = CaseStore::open(tmp.path().join("cases")).await.unwrap();
let store = Arc::new(store_inner);
let cache = Arc::new(SnapshotCache::new(tmp.path().join("cache.json")));
let poller = OnelinerPoller::spawn(
Arc::new(reqwest::Client::new()),
PollerConfig {
server_url: server.uri(),
api_key: TEST_KEY.into(),
poll_interval: Duration::from_millis(50),
window_hours: 72,
},
store.clone(),
cache,
);
// Let the immediate first poll fire.
tokio::time::sleep(Duration::from_millis(200)).await;
assert!(poller.last_success().is_some(), "expected at least one poll");
assert_eq!(store.list().await.len(), 1);
// Drop aborts the task; subsequent store inspection must not
// race with a concurrent merge — asserting no panic is enough.
drop(poller);
tokio::time::sleep(Duration::from_millis(50)).await;
}
#[tokio::test]
async fn spawn_with_matching_mock_requires_api_key_header() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path(ONELINERS_PATH))
.and(header_exists(API_KEY_HEADER))
.respond_with(
ResponseTemplate::new(200)
.insert_header("etag", TEST_ETAG)
.set_body_json(response_body(
"550e8400-e29b-41d4-a716-000000000005",
None,
)),
)
.expect(1)
.mount(&server)
.await;
let tmp = TempDir::new().unwrap();
let (store_inner, _rx) = CaseStore::open(tmp.path().join("cases")).await.unwrap();
let store = Arc::new(store_inner);
let cache = Arc::new(SnapshotCache::new(tmp.path().join("cache.json")));
let _poller = OnelinerPoller::spawn(
Arc::new(reqwest::Client::new()),
PollerConfig {
server_url: server.uri(),
api_key: TEST_KEY.into(),
poll_interval: Duration::from_secs(10),
window_hours: 72,
},
store.clone(),
cache,
);
tokio::time::sleep(Duration::from_millis(200)).await;
// Mock's `.expect(1)` is verified when `server` drops at end of test.
}
}