linkrs_server/http/handlers/
sync.rs

1use axum::{
2    extract::{Extension, Query, State},
3    response::Json as JsonResponse,
4    Json,
5};
6use serde::{Deserialize, Serialize};
7
8use crate::http::{error::HttpError, state::AppState};
9use crate::storage::{
10    StorageClient, StorageOperationContextOps, StorageSchemaContextOps, StorageSyncContextOps,
11};
12
13/// Sync status response
14#[derive(Debug, Serialize, utoipa::ToSchema)]
15pub struct SyncStatusResponse {
16    pub is_running: bool,
17    pub dlq_size: usize,
18    pub unrecovered_dlq_size: usize,
19    pub outbox_pending: usize,
20    pub outbox_retries: u64,
21    pub outbox_dead_lettered: usize,
22    pub outbox_leased: usize,
23    pub outbox_oldest_event_age_ms: u64,
24    pub outbox_write_amplification_bytes: u64,
25    pub outbox_lock_wait_nanos: u64,
26    pub outbox_persist_operations: u64,
27}
28
29#[utoipa::path(
30    get,
31    operation_id = "get_v1_sync_status",
32    path = "/v1/sync/status",
33    tag = "Sync",
34    responses(
35        (status = 200, body = SyncStatusResponse, description = "Sync status"),
36        (status = 500, description = "Internal error")
37    )
38)]
39/// Get sync status
40pub async fn status<
41    S: StorageClient
42        + StorageSchemaContextOps
43        + StorageSyncContextOps
44        + StorageOperationContextOps
45        + Clone
46        + Send
47        + Sync
48        + 'static,
49>(
50    State(state): State<AppState<S>>,
51    Extension(session_id): Extension<i64>,
52) -> Result<JsonResponse<SyncStatusResponse>, HttpError> {
53    super::authz::require_admin_session(&state, session_id)?;
54    let graph_service = state.server.get_graph_service();
55    let sync_api = graph_service.sync_api();
56
57    if let Some(sync_api) = sync_api {
58        let outbox = sync_api.outbox_stats();
59        Ok(JsonResponse(SyncStatusResponse {
60            is_running: sync_api.is_running(),
61            dlq_size: sync_api.get_dlq_size(),
62            unrecovered_dlq_size: sync_api.get_unrecovered_dlq_size(),
63            outbox_pending: outbox.pending,
64            outbox_retries: outbox.retries,
65            outbox_dead_lettered: outbox.dead_lettered,
66            outbox_leased: outbox.leased,
67            outbox_oldest_event_age_ms: outbox.oldest_event_age_ms,
68            outbox_write_amplification_bytes: outbox.write_amplification_bytes,
69            outbox_lock_wait_nanos: outbox.lock_wait_nanos,
70            outbox_persist_operations: outbox.persist_operations,
71        }))
72    } else {
73        // Sync manager not available, return disabled status
74        Ok(JsonResponse(SyncStatusResponse {
75            is_running: false,
76            dlq_size: 0,
77            unrecovered_dlq_size: 0,
78            outbox_pending: 0,
79            outbox_retries: 0,
80            outbox_dead_lettered: 0,
81            outbox_leased: 0,
82            outbox_oldest_event_age_ms: 0,
83            outbox_write_amplification_bytes: 0,
84            outbox_lock_wait_nanos: 0,
85            outbox_persist_operations: 0,
86        }))
87    }
88}
89
90#[utoipa::path(
91    post,
92    path = "/v1/sync/outbox/retry",
93    tag = "Sync",
94    responses(
95        (status = 200, body = serde_json::Value, description = "Outbox delivery retried"),
96        (status = 500, description = "Internal error")
97    )
98)]
99/// Retry delivery of pending durable outbox entries.
100pub async fn retry_outbox<
101    S: StorageClient
102        + StorageSchemaContextOps
103        + StorageSyncContextOps
104        + StorageOperationContextOps
105        + Clone
106        + Send
107        + Sync
108        + 'static,
109>(
110    State(state): State<AppState<S>>,
111    Extension(session_id): Extension<i64>,
112) -> Result<JsonResponse<serde_json::Value>, HttpError> {
113    super::authz::require_admin_session(&state, session_id)?;
114    let graph_service = state.server.get_graph_service();
115    let sync_api = graph_service
116        .sync_api()
117        .ok_or_else(|| HttpError::bad_request("Synchronization is not configured"))?;
118    let delivered = sync_api
119        .retry_outbox_projection()
120        .map_err(HttpError::from)?;
121    Ok(JsonResponse(serde_json::json!({
122        "delivered": delivered,
123        "status": "completed",
124    })))
125}
126
127#[utoipa::path(
128    get,
129    path = "/v1/sync/outbox/diagnostics",
130    tag = "Sync",
131    responses(
132        (status = 200, body = serde_json::Value, description = "Outbox diagnostics"),
133        (status = 500, description = "Internal error")
134    )
135)]
136/// Diagnostics query for outbox frontiers and degraded state.
137pub async fn diagnostics<
138    S: StorageClient
139        + StorageSchemaContextOps
140        + StorageSyncContextOps
141        + StorageOperationContextOps
142        + Clone
143        + Send
144        + Sync
145        + 'static,
146>(
147    State(state): State<AppState<S>>,
148    Extension(session_id): Extension<i64>,
149) -> Result<JsonResponse<serde_json::Value>, HttpError> {
150    super::authz::require_admin_session(&state, session_id)?;
151    let graph_service = state.server.get_graph_service();
152    let sync_api = graph_service
153        .sync_api()
154        .ok_or_else(|| HttpError::bad_request("Synchronization is not configured"))?;
155    let diag = sync_api.sync_diagnostics().map_err(HttpError::from)?;
156    Ok(JsonResponse(serde_json::json!(diag)))
157}
158
159#[derive(Debug, Deserialize, utoipa::ToSchema)]
160pub struct DeadLetterQuery {
161    pub target: Option<String>,
162    pub index_id: Option<u64>,
163    pub generation: Option<u64>,
164    pub limit: Option<usize>,
165    pub offset: Option<usize>,
166}
167
168#[utoipa::path(
169    get,
170    path = "/v1/sync/outbox/dead_letters",
171    tag = "Sync",
172    params(
173        ("target" = Option<String>, Query, description = "Sync target filter"),
174        ("index_id" = Option<u64>, Query, description = "Index id filter"),
175        ("generation" = Option<u64>, Query, description = "Generation filter"),
176        ("limit" = Option<usize>, Query, description = "Maximum rows returned"),
177        ("offset" = Option<usize>, Query, description = "Rows to skip")
178    ),
179    responses(
180        (status = 200, body = serde_json::Value, description = "Dead letter entries"),
181        (status = 500, description = "Internal error")
182    )
183)]
184pub async fn dead_letters<
185    S: StorageClient
186        + StorageSchemaContextOps
187        + StorageSyncContextOps
188        + StorageOperationContextOps
189        + Clone
190        + Send
191        + Sync
192        + 'static,
193>(
194    State(state): State<AppState<S>>,
195    Extension(session_id): Extension<i64>,
196    Query(params): Query<DeadLetterQuery>,
197) -> Result<JsonResponse<serde_json::Value>, HttpError> {
198    super::authz::require_admin_session(&state, session_id)?;
199    let graph_service = state.server.get_graph_service();
200    let sync_api = graph_service
201        .sync_api()
202        .ok_or_else(|| HttpError::bad_request("Synchronization is not configured"))?;
203    let target = params
204        .target
205        .as_deref()
206        .map(|s| linkrs_core::types::TargetId::new(s.to_string()).map_err(HttpError::bad_request))
207        .transpose()?;
208    let rows = sync_api
209        .list_dead_letters(
210            target.as_ref(),
211            params.index_id,
212            params.generation,
213            params.limit.unwrap_or(100),
214            params.offset.unwrap_or(0),
215        )
216        .map_err(HttpError::from)?;
217    Ok(JsonResponse(serde_json::json!({ "dead_letters": rows })))
218}
219
220#[derive(Debug, Deserialize, Serialize, utoipa::ToSchema)]
221pub struct RequeueRequest {
222    pub target: Option<String>,
223    pub index_id: Option<u64>,
224    pub generation: Option<u64>,
225    pub limit: Option<usize>,
226    pub event_ids: Option<Vec<i64>>,
227}
228
229#[utoipa::path(
230    post,
231    path = "/v1/sync/outbox/requeue",
232    tag = "Sync",
233    request_body = RequeueRequest,
234    responses(
235        (status = 200, body = serde_json::Value, description = "Dead letters requeued"),
236        (status = 500, description = "Internal error")
237    )
238)]
239pub async fn requeue<
240    S: StorageClient
241        + StorageSchemaContextOps
242        + StorageSyncContextOps
243        + StorageOperationContextOps
244        + Clone
245        + Send
246        + Sync
247        + 'static,
248>(
249    State(state): State<AppState<S>>,
250    Extension(session_id): Extension<i64>,
251    Json(payload): Json<RequeueRequest>,
252) -> Result<JsonResponse<serde_json::Value>, HttpError> {
253    super::authz::require_admin_session(&state, session_id)?;
254    let graph_service = state.server.get_graph_service();
255    let sync_api = graph_service
256        .sync_api()
257        .ok_or_else(|| HttpError::bad_request("Synchronization is not configured"))?;
258    if let Some(ids) = payload.event_ids {
259        let mut requeued = 0usize;
260        for id in ids {
261            if sync_api.requeue_dead_letter(id).map_err(HttpError::from)? {
262                requeued += 1;
263            }
264        }
265        return Ok(JsonResponse(serde_json::json!({ "requeued": requeued })));
266    }
267    let target = payload
268        .target
269        .as_deref()
270        .map(|s| linkrs_core::types::TargetId::new(s.to_string()).map_err(HttpError::bad_request))
271        .transpose()?;
272    let requeued = sync_api
273        .requeue_dead_letters_batch(
274            target.as_ref(),
275            payload.index_id,
276            payload.generation,
277            payload.limit.unwrap_or(100),
278        )
279        .map_err(HttpError::from)?;
280    Ok(JsonResponse(serde_json::json!({ "requeued": requeued })))
281}
282
283#[derive(Debug, Deserialize, utoipa::ToSchema)]
284pub struct DegradedQuery {
285    pub target: Option<String>,
286    pub index_id: Option<u64>,
287    pub generation: Option<u64>,
288}
289
290#[utoipa::path(
291    get,
292    path = "/v1/sync/outbox/degraded_ranges",
293    tag = "Sync",
294    params(
295        ("target" = Option<String>, Query, description = "Sync target filter"),
296        ("index_id" = Option<u64>, Query, description = "Index id filter"),
297        ("generation" = Option<u64>, Query, description = "Generation filter")
298    ),
299    responses(
300        (status = 200, body = serde_json::Value, description = "Degraded ranges"),
301        (status = 500, description = "Internal error")
302    )
303)]
304pub async fn degraded_ranges<
305    S: StorageClient
306        + StorageSchemaContextOps
307        + StorageSyncContextOps
308        + StorageOperationContextOps
309        + Clone
310        + Send
311        + Sync
312        + 'static,
313>(
314    State(state): State<AppState<S>>,
315    Extension(session_id): Extension<i64>,
316    Query(params): Query<DegradedQuery>,
317) -> Result<JsonResponse<serde_json::Value>, HttpError> {
318    super::authz::require_admin_session(&state, session_id)?;
319    let graph_service = state.server.get_graph_service();
320    let sync_api = graph_service
321        .sync_api()
322        .ok_or_else(|| HttpError::bad_request("Synchronization is not configured"))?;
323    let target = params
324        .target
325        .as_deref()
326        .map(|s| linkrs_core::types::TargetId::new(s.to_string()).map_err(HttpError::bad_request))
327        .transpose()?;
328    let rows = sync_api
329        .list_degraded_ranges(target.as_ref(), params.index_id, params.generation)
330        .map_err(HttpError::from)?;
331    Ok(JsonResponse(serde_json::json!({ "degraded_ranges": rows })))
332}
333
334#[derive(Debug, Deserialize, utoipa::ToSchema)]
335pub struct ClearDegradedRequest {
336    pub target: String,
337    pub index_id: u64,
338    pub generation: u64,
339    pub start_lsn: u64,
340    pub end_lsn: u64,
341}
342
343#[utoipa::path(
344    post,
345    path = "/v1/sync/outbox/degraded/clear",
346    tag = "Sync",
347    request_body = ClearDegradedRequest,
348    responses(
349        (status = 200, body = serde_json::Value, description = "Degraded range cleared"),
350        (status = 500, description = "Internal error")
351    )
352)]
353pub async fn degraded_clear<
354    S: StorageClient
355        + StorageSchemaContextOps
356        + StorageSyncContextOps
357        + StorageOperationContextOps
358        + Clone
359        + Send
360        + Sync
361        + 'static,
362>(
363    State(state): State<AppState<S>>,
364    Extension(session_id): Extension<i64>,
365    Json(payload): Json<ClearDegradedRequest>,
366) -> Result<JsonResponse<serde_json::Value>, HttpError> {
367    super::authz::require_admin_session(&state, session_id)?;
368    let graph_service = state.server.get_graph_service();
369    let sync_api = graph_service
370        .sync_api()
371        .ok_or_else(|| HttpError::bad_request("Synchronization is not configured"))?;
372    let target =
373        linkrs_core::types::TargetId::new(payload.target).map_err(HttpError::bad_request)?;
374    let cleared = sync_api
375        .clear_degraded_range(
376            &target,
377            payload.index_id,
378            payload.generation,
379            linkrs_core::types::CommitLsn::new(payload.start_lsn),
380            linkrs_core::types::CommitLsn::new(payload.end_lsn),
381        )
382        .map_err(HttpError::from)?;
383    Ok(JsonResponse(serde_json::json!({ "cleared": cleared })))
384}
385
386#[derive(Debug, Deserialize, utoipa::ToSchema)]
387pub struct RetentionRunRequest {
388    pub grace_lsn_distance: Option<u64>,
389    pub max_age_ms: Option<u64>,
390}
391
392#[utoipa::path(
393    post,
394    path = "/v1/sync/outbox/retention/run",
395    tag = "Sync",
396    request_body = RetentionRunRequest,
397    responses(
398        (status = 200, body = serde_json::Value, description = "Retention pass completed"),
399        (status = 500, description = "Internal error")
400    )
401)]
402pub async fn retention_run<
403    S: StorageClient
404        + StorageSchemaContextOps
405        + StorageSyncContextOps
406        + StorageOperationContextOps
407        + Clone
408        + Send
409        + Sync
410        + 'static,
411>(
412    State(state): State<AppState<S>>,
413    Extension(session_id): Extension<i64>,
414    Json(payload): Json<RetentionRunRequest>,
415) -> Result<JsonResponse<serde_json::Value>, HttpError> {
416    super::authz::require_admin_session(&state, session_id)?;
417    let graph_service = state.server.get_graph_service();
418    let sync_api = graph_service
419        .sync_api()
420        .ok_or_else(|| HttpError::bad_request("Synchronization is not configured"))?;
421    let grace = payload.grace_lsn_distance.unwrap_or(10_000);
422    let max_age = payload.max_age_ms.unwrap_or(86_400_000 * 30);
423    let (pruned, archived, retention_lsn) = sync_api
424        .run_retention_once(grace, max_age)
425        .map_err(HttpError::from)?;
426    Ok(JsonResponse(serde_json::json!({
427        "pruned": pruned,
428        "archived": archived,
429        "retention_lsn": retention_lsn,
430    })))
431}
432
433#[utoipa::path(
434    get,
435    path = "/v1/sync/outbox/retention/status",
436    tag = "Sync",
437    responses(
438        (status = 200, body = serde_json::Value, description = "Retention watermark"),
439        (status = 500, description = "Internal error")
440    )
441)]
442pub async fn retention_status<
443    S: StorageClient
444        + StorageSchemaContextOps
445        + StorageSyncContextOps
446        + StorageOperationContextOps
447        + Clone
448        + Send
449        + Sync
450        + 'static,
451>(
452    State(state): State<AppState<S>>,
453    Extension(session_id): Extension<i64>,
454) -> Result<JsonResponse<serde_json::Value>, HttpError> {
455    super::authz::require_admin_session(&state, session_id)?;
456    let graph_service = state.server.get_graph_service();
457    let sync_api = graph_service
458        .sync_api()
459        .ok_or_else(|| HttpError::bad_request("Synchronization is not configured"))?;
460    let retention = sync_api.retention_lsn().map_err(HttpError::from)?;
461    Ok(JsonResponse(
462        serde_json::json!({ "retention_lsn": retention.get() }),
463    ))
464}
linkrs_server/http/handlers/sync.rs

linkrs_server/http/handlers/
sync.rs
