diff --git a/frontend/src/components/config/config-editor.tsx b/frontend/src/components/config/config-editor.tsx index b665673..a72b7e2 100644 --- a/frontend/src/components/config/config-editor.tsx +++ b/frontend/src/components/config/config-editor.tsx @@ -1,4 +1,4 @@ -import { useBlocker, useNavigate } from "@tanstack/react-router"; +import { useBlocker } from "@tanstack/react-router"; import { AlertTriangle, CheckCircle2, @@ -30,11 +30,12 @@ import { DialogTitle, } from "#/components/ui/dialog"; import { WorkspaceLayout } from "#/components/ui/workspace-layout"; -import { useAuth } from "#/lib/auth"; +import { ApiRequestError } from "#/lib/api"; import { type ConfigDocument, type ConfigWarning, loadConfigDocument, + loadServiceStatus, scheduleRestart, } from "#/lib/config-api"; @@ -118,13 +119,11 @@ function ConfigWorkspace({ backgroundError, }: ConfigWorkspaceProps) { const draft = useConfigDraft(initialDocument); - const { setAuth } = useAuth(); - const navigate = useNavigate(); const [mobileNavigationOpen, setMobileNavigationOpen] = useState(false); const [restartOpen, setRestartOpen] = useState(false); const [restartBusy, setRestartBusy] = useState(false); const [actionMessage, setActionMessage] = useState(""); - const allowNavigation = useRef(false); + const restartMonitorGeneration = useRef(0); const { t } = useTranslation(); const activeSection = useMemo( () => @@ -136,7 +135,6 @@ function ConfigWorkspace({ const blocker = useBlocker({ shouldBlockFn: ({ current, next }) => - !allowNavigation.current && draft.isDirty && current.pathname === "/config" && next.pathname !== "/config", @@ -145,29 +143,59 @@ function ConfigWorkspace({ withResolver: true, }); - async function handleConfirmSave() { - const result = await draft.confirmSave(); - if (!result) return; - if (result.session_invalidated) { - allowNavigation.current = true; + useEffect( + () => () => { + restartMonitorGeneration.current += 1; + }, + [], + ); + + async function monitorRestart() { + const generation = restartMonitorGeneration.current + 1; + restartMonitorGeneration.current = generation; + for (let attempt = 0; attempt < 60; attempt += 1) { + if (attempt > 0) { + await new Promise((resolve) => window.setTimeout(resolve, 500)); + } + if (restartMonitorGeneration.current !== generation) return; try { - setAuth({ authenticated: false }); - await navigate({ - to: "/login", - search: { - notice: "config_saved_restart_scheduled", - }, - }); - } finally { - allowNavigation.current = false; + const status = await loadServiceStatus(); + if (restartMonitorGeneration.current !== generation) return; + if (status.restart_status === "command_failed") { + setActionMessage(t("config.status.restartCommandFailed")); + return; + } + if ( + status.restart_status === "command_completed" || + status.restart_status === "idle" + ) { + return; + } + } catch (statusError) { + // A 401 is handled globally and proves that a new process with changed + // credentials is serving requests. Other failures are inconclusive. + if ( + statusError instanceof ApiRequestError && + statusError.status === 401 + ) { + return; + } } - return; } + if (restartMonitorGeneration.current === generation) { + setActionMessage(t("config.status.restartStatusUnavailable")); + } + } + + async function handleConfirmSave() { + const result = await draft.confirmSave(); + if (!result) return; setActionMessage( result.requires_restart ? t("config.status.savedRestart") : t("config.status.saved"), ); + if (result.restart_scheduled) void monitorRestart(); } async function handleRestart() { @@ -177,6 +205,7 @@ function ConfigWorkspace({ await scheduleRestart(); setActionMessage(t("config.status.restartScheduled")); setRestartOpen(false); + void monitorRestart(); } catch (restartError) { setActionMessage( t("config.status.restartFailed", { diff --git a/frontend/src/components/config/use-config-draft.ts b/frontend/src/components/config/use-config-draft.ts index 499dc5c..452a077 100644 --- a/frontend/src/components/config/use-config-draft.ts +++ b/frontend/src/components/config/use-config-draft.ts @@ -202,9 +202,7 @@ export function useConfigDraft(initialDocument: ConfigDocument) { setBaseline(structuredClone(snapshot)); setDraft(structuredClone(snapshot)); setBaseRevision(result.revision); - setRestartRequired( - result.requires_restart && !result.restart_scheduled, - ); + setRestartRequired(result.requires_restart); draftVersion.current += 1; setCheck({ status: "idle" }); setPreview({ status: "closed" }); diff --git a/frontend/src/config-editor.test.tsx b/frontend/src/config-editor.test.tsx index d1862b0..661d968 100644 --- a/frontend/src/config-editor.test.tsx +++ b/frontend/src/config-editor.test.tsx @@ -104,6 +104,14 @@ type PreviewOverrides = Partial<{ hasChanges: boolean; passwordChange: boolean; warnings: string[]; + restartStatus: "idle" | "scheduled" | "command_completed" | "command_failed"; + restartStatusSequence: Array< + | "idle" + | "scheduled" + | "command_completed" + | "command_failed" + | "network_error" + >; }>; function previewResponse(previewOverrides: PreviewOverrides = {}) { @@ -140,6 +148,9 @@ function installApi( nextPreview?: () => Promise, ) { const requests: Array<{ url: string; init?: RequestInit }> = []; + const restartStatusSequence = [ + ...(previewOverrides.restartStatusSequence ?? []), + ]; const fetchMock = vi.fn( async (input: RequestInfo | URL, init?: RequestInit) => { const url = String(input); @@ -169,7 +180,25 @@ function installApi( revision: "saved-revision", requires_restart: true, restart_scheduled: url.includes("restart_after_save=true"), - session_invalidated: url.includes("restart_after_save=true"), + session_invalidated: false, + }), + { status: 200, headers: { "Content-Type": "application/json" } }, + ); + } + if (url === "/api/status") { + const nextStatus = restartStatusSequence.shift(); + if (nextStatus === "network_error") { + throw new Error("temporary connection failure"); + } + return new Response( + JSON.stringify({ + version: "test", + uptime_seconds: 1, + api_bind: "0.0.0.0", + api_port: 8080, + database_path: "/tmp/sms-relayed.sqlite", + restart_status: + nextStatus ?? previewOverrides.restartStatus ?? "scheduled", }), { status: 200, headers: { "Content-Type": "application/json" } }, ); @@ -530,32 +559,25 @@ describe("ConfigEditor workspace", () => { await screen.findByText("Check failed: invalid device name"); }); - test("uses combined save and restart for a password change then signs out", async () => { + test("keeps the current session when a password-change restart command fails", async () => { const setAuth = vi.fn(); - const leavingConfig = { - current: { pathname: "/config" }, - next: { pathname: "/login" }, - }; const { requests } = installApi({ passwordChange: true, warnings: ["password_change"], + restartStatus: "command_failed", }); render(); fireEvent.change(await screen.findByLabelText("Password"), { target: { value: "new-password" }, }); - expect(routerMocks.shouldBlockFn(leavingConfig)).toBe(true); - routerMocks.navigate.mockImplementation(async () => { - expect(routerMocks.shouldBlockFn(leavingConfig)).toBe(false); - }); fireEvent.click(screen.getByRole("button", { name: "Save" })); fireEvent.click( await screen.findByRole("button", { name: "Save and schedule restart" }), ); - await waitFor(() => - expect(setAuth).toHaveBeenCalledWith({ authenticated: false }), + await screen.findByText( + "Service restart command failed. Your current session is still active; retry restart.", ); const saveRequest = requests.find( (request) => request.init?.method === "PUT", @@ -564,12 +586,73 @@ describe("ConfigEditor workspace", () => { expect( requests.some((request) => request.url === "/api/service/restart"), ).toBe(false); - expect(routerMocks.navigate).toHaveBeenCalledWith( - expect.objectContaining({ - to: "/login", - search: { notice: "config_saved_restart_scheduled" }, - }), + expect(requests.some((request) => request.url === "/api/status")).toBe( + true, + ); + expect(setAuth).not.toHaveBeenCalled(); + expect(routerMocks.navigate).not.toHaveBeenCalled(); + expect(screen.getByText(/Restart required/)).toBeTruthy(); + }); + + test("continues monitoring after a transient restart status failure", async () => { + const setAuth = vi.fn(); + const { requests } = installApi({ + passwordChange: true, + warnings: ["password_change"], + restartStatusSequence: ["network_error", "command_failed"], + }); + render(); + + fireEvent.change(await screen.findByLabelText("Password"), { + target: { value: "new-password" }, + }); + fireEvent.click(screen.getByRole("button", { name: "Save" })); + fireEvent.click( + await screen.findByRole("button", { name: "Save and schedule restart" }), + ); + + await screen.findByText( + "Service restart command failed. Your current session is still active; retry restart.", ); + expect( + requests.filter((request) => request.url === "/api/status"), + ).toHaveLength(2); + expect(setAuth).not.toHaveBeenCalled(); + expect(routerMocks.navigate).not.toHaveBeenCalled(); + }); + + test("keeps the success message after the restart command completes", async () => { + installApi({ + passwordChange: true, + warnings: ["password_change"], + restartStatusSequence: ["command_completed", "command_failed"], + }); + render(); + + fireEvent.change(await screen.findByLabelText("Password"), { + target: { value: "new-password" }, + }); + fireEvent.click(screen.getByRole("button", { name: "Save" })); + fireEvent.click( + await screen.findByRole("button", { name: "Save and schedule restart" }), + ); + + await screen.findByText("Configuration saved. Restart required."); + await act( + () => + new Promise((resolve) => { + window.setTimeout(resolve, 600); + }), + ); + + expect( + screen.getByText("Configuration saved. Restart required."), + ).toBeTruthy(); + expect( + screen.queryByText( + "Service restart command failed. Your current session is still active; retry restart.", + ), + ).toBeNull(); }); test("uses singular category copy for one changed section", async () => { diff --git a/frontend/src/lib/config-api.ts b/frontend/src/lib/config-api.ts index 2e495b1..9f975ee 100644 --- a/frontend/src/lib/config-api.ts +++ b/frontend/src/lib/config-api.ts @@ -1,5 +1,5 @@ import { apiFetch, apiRequest } from "#/lib/api"; -import type { AppConfig } from "#/lib/config-model"; +import type { AppConfig, StatusResponse } from "#/lib/config-model"; export type ConfigDocument = { config: AppConfig; @@ -91,3 +91,7 @@ export async function saveConfig( export async function scheduleRestart(): Promise { await apiFetch("/api/service/restart", { method: "POST" }); } + +export async function loadServiceStatus(): Promise { + return apiFetch("/api/status"); +} diff --git a/frontend/src/lib/config-model.ts b/frontend/src/lib/config-model.ts index bd20686..36fb2c5 100644 --- a/frontend/src/lib/config-model.ts +++ b/frontend/src/lib/config-model.ts @@ -65,4 +65,5 @@ export type StatusResponse = { api_bind: string; api_port: number; database_path: string; + restart_status: "idle" | "scheduled" | "command_completed" | "command_failed"; }; diff --git a/frontend/src/locales/en.ts b/frontend/src/locales/en.ts index ebc1f90..4b8f718 100644 --- a/frontend/src/locales/en.ts +++ b/frontend/src/locales/en.ts @@ -453,6 +453,10 @@ export const en = { restartScheduled: "Restart scheduled. The dashboard may disconnect briefly.", restartFailed: "Restart failed: {{message}}", + restartCommandFailed: + "Service restart command failed. Your current session is still active; retry restart.", + restartStatusUnavailable: + "Unable to confirm the service restart. Your current session is still active; retry restart.", }, restartDialog: { title: "Schedule service restart?", @@ -487,7 +491,7 @@ export const en = { }, warnings: { passwordChange: - "All sessions will be signed out after Save + Restart is scheduled.", + "Current sessions stay active until the service restarts, then all sessions are signed out.", apiDisable: "The dashboard will be unavailable after restart.", apiEndpointChange: "The dashboard address may change after restart.", trustedProxiesChange: @@ -564,7 +568,7 @@ export const en = { "Also listen on a safe IPv6 companion address when one can be inferred.", password: "Password", passwordDescription: - "Changing this value saves and schedules restart in one step, then signs out every session.", + "Changing this value saves and schedules restart in one step. Sessions are signed out only after the new service starts.", databasePath: "Database path", }, timeouts: { diff --git a/frontend/src/locales/es.ts b/frontend/src/locales/es.ts index 28097d8..04c8284 100644 --- a/frontend/src/locales/es.ts +++ b/frontend/src/locales/es.ts @@ -471,6 +471,10 @@ export const es = { restartScheduled: "Reinicio programado. El panel puede desconectarse brevemente.", restartFailed: "Reinicio fallido: {{message}}", + restartCommandFailed: + "El comando de reinicio del servicio falló. La sesión actual sigue activa; vuelve a intentar el reinicio.", + restartStatusUnavailable: + "No se pudo confirmar el reinicio del servicio. La sesión actual sigue activa; vuelve a intentarlo.", }, restartDialog: { title: "¿Programar reinicio del servicio?", @@ -505,7 +509,7 @@ export const es = { }, warnings: { passwordChange: - "Todas las sesiones se cerrarán tras programar Guardar + Reiniciar.", + "Las sesiones actuales siguen activas hasta que el servicio se reinicie; después se cerrarán todas.", apiDisable: "El panel no estará disponible tras el reinicio.", apiEndpointChange: "La dirección del panel puede cambiar tras el reinicio.", @@ -583,7 +587,7 @@ export const es = { "Escuchar también en una dirección IPv6 complementaria segura cuando se pueda inferir una.", password: "Contraseña", passwordDescription: - "Cambiar este valor guarda y programa el reinicio en un solo paso y luego cierra todas las sesiones.", + "Cambiar este valor guarda y programa el reinicio en un solo paso. Las sesiones se cierran solo cuando inicia el nuevo servicio.", databasePath: "Ruta de la base de datos", }, timeouts: { diff --git a/frontend/src/locales/fr.ts b/frontend/src/locales/fr.ts index 25ca04b..3e3531b 100644 --- a/frontend/src/locales/fr.ts +++ b/frontend/src/locales/fr.ts @@ -468,6 +468,10 @@ export const fr = { restartScheduled: "Redémarrage planifié. Le tableau de bord peut se déconnecter brièvement.", restartFailed: "Échec du redémarrage : {{message}}", + restartCommandFailed: + "La commande de redémarrage du service a échoué. Votre session actuelle reste active ; réessayez le redémarrage.", + restartStatusUnavailable: + "Impossible de confirmer le redémarrage du service. Votre session actuelle reste active ; réessayez.", }, restartDialog: { title: "Planifier le redémarrage du service ?", @@ -502,7 +506,7 @@ export const fr = { }, warnings: { passwordChange: - "Toutes les sessions seront déconnectées après l'enregistrement et la planification du redémarrage.", + "Les sessions actuelles restent actives jusqu'au redémarrage du service, puis elles sont toutes déconnectées.", apiDisable: "Le tableau de bord sera indisponible après le redémarrage.", apiEndpointChange: "L'adresse du tableau de bord peut changer après le redémarrage.", @@ -580,7 +584,7 @@ export const fr = { "Écoute également sur une adresse compagnon IPv6 sûre lorsqu'elle peut être déduite.", password: "Mot de passe", passwordDescription: - "Modifier cette valeur enregistre et planifie le redémarrage en une seule étape, puis déconnecte toutes les sessions.", + "Modifier cette valeur enregistre et planifie le redémarrage en une seule étape. Les sessions ne sont déconnectées qu'au démarrage du nouveau service.", databasePath: "Chemin de la base de données", }, timeouts: { diff --git a/frontend/src/locales/ja.ts b/frontend/src/locales/ja.ts index 6f5214d..dc8e553 100644 --- a/frontend/src/locales/ja.ts +++ b/frontend/src/locales/ja.ts @@ -457,6 +457,10 @@ export const ja = { restartScheduled: "再起動がスケジュールされました。ダッシュボードが一時的に切断される場合があります。", restartFailed: "再起動に失敗しました：{{message}}", + restartCommandFailed: + "サービスの再起動コマンドに失敗しました。現在のセッションは有効なままです。再起動を再試行してください。", + restartStatusUnavailable: + "サービスの再起動を確認できません。現在のセッションは有効なままです。再起動を再試行してください。", }, restartDialog: { title: "サービスの再起動をスケジュールしますか？", @@ -491,7 +495,7 @@ export const ja = { }, warnings: { passwordChange: - "保存と再起動のスケジュール後、すべてのセッションからサインアウトされます。", + "サービスが再起動するまで現在のセッションは有効です。新しいサービスの起動後にすべてのセッションが無効になります。", apiDisable: "再起動後、ダッシュボードは利用できなくなります。", apiEndpointChange: "再起動後、ダッシュボードのアドレスが変更される場合があります。", @@ -568,7 +572,7 @@ export const ja = { "安全なIPv6コンパニオンアドレスを推測できる場合にも、そのアドレスでリッスンします。", password: "パスワード", passwordDescription: - "この値を変更すると、保存と再起動のスケジュールが一度に行われ、すべてのセッションからサインアウトされます。", + "この値を変更すると、保存と再起動のスケジュールが一度に行われます。新しいサービスの起動後にのみ、すべてのセッションが無効になります。", databasePath: "データベースパス", }, timeouts: { diff --git a/frontend/src/locales/ko.ts b/frontend/src/locales/ko.ts index 5debafd..b6b90a8 100644 --- a/frontend/src/locales/ko.ts +++ b/frontend/src/locales/ko.ts @@ -452,6 +452,10 @@ export const ko = { restartScheduled: "재시작이 예약되었습니다. 대시보드가 잠시 끊길 수 있습니다.", restartFailed: "재시작 실패: {{message}}", + restartCommandFailed: + "서비스 재시작 명령이 실패했습니다. 현재 세션은 계속 유효합니다. 재시작을 다시 시도하세요.", + restartStatusUnavailable: + "서비스 재시작을 확인할 수 없습니다. 현재 세션은 계속 유효합니다. 재시작을 다시 시도하세요.", }, restartDialog: { title: "서비스 재시작을 예약하시겠습니까?", @@ -485,7 +489,8 @@ export const ko = { saveAndRestart: "저장 및 재시작 예약", }, warnings: { - passwordChange: "저장 및 재시작 예약 후 모든 세션에서 로그아웃됩니다.", + passwordChange: + "서비스가 재시작될 때까지 현재 세션은 유지되며, 새 서비스가 시작된 후 모든 세션이 무효화됩니다.", apiDisable: "재시작 후 대시보드를 사용할 수 없습니다.", apiEndpointChange: "재시작 후 대시보드 주소가 변경될 수 있습니다.", trustedProxiesChange: @@ -561,7 +566,7 @@ export const ko = { "안전한 IPv6 동반 주소를 추론할 수 있을 때 해당 주소에서도 수신합니다.", password: "비밀번호", passwordDescription: - "이 값을 변경하면 한 단계로 저장 및 재시작이 예약된 후 모든 세션에서 로그아웃됩니다.", + "이 값을 변경하면 한 단계로 저장 및 재시작이 예약됩니다. 새 서비스가 시작된 후에만 모든 세션이 무효화됩니다.", databasePath: "데이터베이스 경로", }, timeouts: { diff --git a/frontend/src/locales/zh-CN.ts b/frontend/src/locales/zh-CN.ts index 3d4eaa8..dc6d88d 100644 --- a/frontend/src/locales/zh-CN.ts +++ b/frontend/src/locales/zh-CN.ts @@ -434,6 +434,9 @@ export const zhCN = { savedRestart: "配置已保存。需要重启。", restartScheduled: "重启已安排。仪表盘可能短暂断开。", restartFailed: "重启失败：{{message}}", + restartCommandFailed: "服务重启命令失败。当前会话仍然有效；请重试重启。", + restartStatusUnavailable: + "无法确认服务是否已重启。当前会话仍然有效；请重试重启。", }, restartDialog: { title: "安排服务重启？", @@ -466,7 +469,8 @@ export const zhCN = { saveAndRestart: "保存并安排重启", }, warnings: { - passwordChange: "保存并安排重启后，所有会话将被登出。", + passwordChange: + "服务重启前当前会话保持有效；新服务启动后所有会话才会失效。", apiDisable: "重启后仪表盘将不可用。", apiEndpointChange: "重启后仪表盘地址可能更改。", trustedProxiesChange: "重启后，登录限流将使用更新后的可信代理列表。", @@ -533,7 +537,8 @@ export const zhCN = { ipv6: "IPv6 辅助地址", ipv6Description: "当可以推断出安全的 IPv6 辅助地址时，也监听该地址。", password: "密码", - passwordDescription: "更改此值将一步保存并安排重启，然后登出所有会话。", + passwordDescription: + "更改此值将一步保存并安排重启。只有新服务启动后，所有会话才会失效。", databasePath: "数据库路径", }, timeouts: { diff --git a/src/api/auth.rs b/src/api/auth.rs index a0ca496..5a1139f 100644 --- a/src/api/auth.rs +++ b/src/api/auth.rs @@ -3,8 +3,6 @@ use std::fs; use std::io::Write; use std::net::IpAddr; use std::path::{Path, PathBuf}; -#[cfg(test)] -use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::time::{Duration as StdDuration, Instant}; @@ -48,8 +46,6 @@ pub struct SessionStore { store: Store, password: Arc, login_failures: Arc>>, - #[cfg(test)] - invalidate_all_failure: Arc, } impl SessionStore { @@ -69,8 +65,6 @@ impl SessionStore { store, password: Arc::from(password), login_failures: Arc::new(Mutex::new(HashMap::new())), - #[cfg(test)] - invalidate_all_failure: Arc::new(AtomicBool::new(false)), } } @@ -105,19 +99,6 @@ impl SessionStore { .await } - pub async fn invalidate_all(&self) -> anyhow::Result<()> { - #[cfg(test)] - if self.invalidate_all_failure.swap(false, Ordering::SeqCst) { - anyhow::bail!("injected session invalidation failure"); - } - self.store.delete_all_auth_sessions().await - } - - #[cfg(test)] - pub(crate) fn fail_next_invalidate_all(&self) { - self.invalidate_all_failure.store(true, Ordering::SeqCst); - } - fn authenticate(&self, peer: IpAddr, password: &str, expected_password: &str) -> LoginResult { let now = Instant::now(); let mut failures = self.login_failures.lock().unwrap(); @@ -544,19 +525,4 @@ mod tests { .unwrap()); let _ = fs::remove_file(credential_secret_path(&config_path)); } - - #[tokio::test] - async fn invalidated_sessions_do_not_return_when_the_password_is_reused() { - let store = crate::persistence::Store::open_in_memory().unwrap(); - let sessions = SessionStore::new(store.clone(), "reused-password"); - let token = sessions.create_session().await.unwrap(); - - sessions.invalidate_all().await.unwrap(); - let sessions_after_password_reuse = SessionStore::new(store, "reused-password"); - - assert!(!sessions_after_password_reuse - .is_valid(&token) - .await - .unwrap()); - } } diff --git a/src/api/config.rs b/src/api/config.rs index b712cd0..0f50ede 100644 --- a/src/api/config.rs +++ b/src/api/config.rs @@ -204,30 +204,20 @@ async fn save_config( .await .map_err(|error| ApiError::internal(error.to_string()))? .map_err(|error| ApiError::internal(error.to_string()))?; - if password_changed { - state - .sessions - .invalidate_all() - .await - .map_err(super::auth::session_storage_error)?; - } tokio::task::spawn_blocking(move || prepared.commit()) .await - .map_err(|error| config_commit_error(error, password_changed))? - .map_err(|error| config_commit_error(error, password_changed))?; + .map_err(|error| ApiError::internal(error.to_string()))? + .map_err(|error| ApiError::internal(error.to_string()))?; } - let restart_scheduled = options.restart_after_save; - if restart_scheduled { - super::service::schedule_restart(&state); - } + let restart_scheduled = options.restart_after_save && super::service::schedule_restart(&state); state.events.send(AppEvent::ConfigSaved); let body = ConfigSaveResponse { revision: candidate_revision.clone(), requires_restart, restart_scheduled, - session_invalidated: password_changed, + session_invalidated: false, }; let mut response = Json(body).into_response(); apply_document_headers( @@ -238,15 +228,6 @@ async fn save_config( Ok(response) } -fn config_commit_error(error: impl std::fmt::Display, sessions_invalidated: bool) -> ApiError { - if sessions_invalidated { - log::error!( - "configuration commit failed after sessions were invalidated for an API password change; authentication state may require operator attention" - ); - } - ApiError::internal(error.to_string()) -} - fn load_config_document_sync(path: &Path) -> anyhow::Result { let content = std::fs::read_to_string(path)?; let config = toml::from_str(&content)?; diff --git a/src/api/mod.rs b/src/api/mod.rs index ff0660d..0e2cf8f 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -492,10 +492,23 @@ mod route_tests { } impl service::ServiceRestarter for RecordingServiceRestarter { - fn restart(&self) { + fn restart(&self) -> anyhow::Result<()> { if let Some(completed) = &self.completed { let _ = completed.send(()); } + Ok(()) + } + } + + #[derive(Clone)] + struct FailingServiceRestarter { + completed: tokio::sync::mpsc::UnboundedSender<()>, + } + + impl service::ServiceRestarter for FailingServiceRestarter { + fn restart(&self) -> anyhow::Result<()> { + let _ = self.completed.send(()); + anyhow::bail!("sensitive restart command details") } } @@ -795,14 +808,72 @@ mod route_tests { } #[tokio::test] - async fn changing_the_api_password_schedules_restart_and_invalidates_sessions() { + async fn restart_failure_is_reported_safely_and_the_current_session_can_retry() { + let mut state = test_state(); + let (completed, mut restarts) = tokio::sync::mpsc::unbounded_channel(); + state.service_control = service::ServiceControl::new(FailingServiceRestarter { completed }); + let service_control = state.service_control.clone(); + let token = state.sessions.create_session().await.unwrap(); + let app = router(state); + + let response = app + .clone() + .oneshot( + Request::builder() + .method(Method::POST) + .uri("/api/service/restart") + .header("cookie", format!("sms-relayed-session={token}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::ACCEPTED); + expect_restart_completed(&mut restarts).await; + expect_restart_idle(&service_control).await; + + let response = app + .clone() + .oneshot( + Request::builder() + .uri("/api/status") + .header("cookie", format!("sms-relayed-session={token}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + let status: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(status["restart_status"], "command_failed"); + assert!(!String::from_utf8_lossy(&body).contains("sensitive restart command details")); + + let retry = app + .oneshot( + Request::builder() + .method(Method::POST) + .uri("/api/service/restart") + .header("cookie", format!("sms-relayed-session={token}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(retry.status(), StatusCode::ACCEPTED); + expect_restart_completed(&mut restarts).await; + } + + #[tokio::test] + async fn changing_the_api_password_schedules_restart_without_invalidating_current_sessions() { let mut state = test_state(); let (config_path, base_revision) = write_config_file(&state.config, "password-change"); state.config_path = config_path.clone(); let (restarter, mut restarts) = RecordingServiceRestarter::with_completion_signal(); state.service_control = service::ServiceControl::new(restarter); let token = state.sessions.create_session().await.unwrap(); - let sessions = state.sessions.clone(); let mut updated_config = (*state.config).clone(); updated_config.api.password = "new-password".to_string(); let candidate_revision = @@ -810,6 +881,7 @@ mod route_tests { let app = router(state); let response = app + .clone() .oneshot( Request::builder() .method(Method::PUT) @@ -832,30 +904,53 @@ mod route_tests { ) .unwrap(); assert_eq!(body["restart_scheduled"], true); - assert_eq!(body["session_invalidated"], true); - assert!(!sessions.is_valid(&token).await.unwrap()); + assert_eq!(body["session_invalidated"], false); + + let authenticated_response = app + .oneshot( + Request::builder() + .uri("/api/status") + .header("cookie", format!("sms-relayed-session={token}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(authenticated_response.status(), StatusCode::OK); expect_restart_completed(&mut restarts).await; let _ = std::fs::remove_file(config_path); } #[tokio::test] - async fn password_change_succeeds_after_rename_when_parent_sync_fails() { + async fn config_save_reports_when_its_restart_request_was_not_accepted() { let mut state = test_state(); let (config_path, base_revision) = - write_config_file(&state.config, "password-parent-sync-failure"); + write_config_file(&state.config, "restart-already-scheduled"); state.config_path = config_path.clone(); let (restarter, mut restarts) = RecordingServiceRestarter::with_completion_signal(); state.service_control = service::ServiceControl::new(restarter); let token = state.sessions.create_session().await.unwrap(); - let sessions = state.sessions.clone(); let mut updated_config = (*state.config).clone(); - updated_config.api.password = "new-password".to_string(); - let candidate_toml = updated_config.canonical_toml().unwrap(); - let candidate_revision = crate::config::config_revision(&candidate_toml); - crate::config::fail_next_config_parent_sync_for(&config_path); + updated_config.app.device_name = "saved-while-restart-pending".to_string(); + let candidate_revision = + crate::config::config_revision(&updated_config.canonical_toml().unwrap()); let app = router(state); + let restart_response = app + .clone() + .oneshot( + Request::builder() + .method(Method::POST) + .uri("/api/service/restart") + .header("cookie", format!("sms-relayed-session={token}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(restart_response.status(), StatusCode::ACCEPTED); + let response = app .oneshot( Request::builder() @@ -872,31 +967,33 @@ mod route_tests { .unwrap(); assert_eq!(response.status(), StatusCode::OK); - assert!(!sessions.is_valid(&token).await.unwrap()); - assert_eq!( - std::fs::read_to_string(&config_path).unwrap(), - candidate_toml - ); + let body: serde_json::Value = serde_json::from_slice( + &axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(), + ) + .unwrap(); + assert_eq!(body["restart_scheduled"], false); expect_restart_completed(&mut restarts).await; let _ = std::fs::remove_file(config_path); } #[tokio::test] - async fn password_change_does_not_schedule_restart_when_session_invalidation_fails() { + async fn password_change_succeeds_after_rename_when_parent_sync_fails() { let mut state = test_state(); let (config_path, base_revision) = - write_config_file(&state.config, "password-invalidation-failure"); + write_config_file(&state.config, "password-parent-sync-failure"); state.config_path = config_path.clone(); - let original = std::fs::read_to_string(&config_path).unwrap(); - state.service_control = service::ServiceControl::new(RecordingServiceRestarter::default()); - let service_control = state.service_control.clone(); + let (restarter, mut restarts) = RecordingServiceRestarter::with_completion_signal(); + state.service_control = service::ServiceControl::new(restarter); let token = state.sessions.create_session().await.unwrap(); - state.sessions.fail_next_invalidate_all(); + let sessions = state.sessions.clone(); let mut updated_config = (*state.config).clone(); updated_config.api.password = "new-password".to_string(); - let candidate_revision = - crate::config::config_revision(&updated_config.canonical_toml().unwrap()); + let candidate_toml = updated_config.canonical_toml().unwrap(); + let candidate_revision = crate::config::config_revision(&candidate_toml); + crate::config::fail_next_config_parent_sync_for(&config_path); let app = router(state); let response = app @@ -906,7 +1003,7 @@ mod route_tests { .uri("/api/config?restart_after_save=true") .header("cookie", format!("sms-relayed-session={token}")) .header("content-type", "application/json") - .header("if-match", base_revision.clone()) + .header("if-match", base_revision) .header("x-config-candidate-revision", candidate_revision) .body(Body::from(serde_json::to_vec(&updated_config).unwrap())) .unwrap(), @@ -914,12 +1011,13 @@ mod route_tests { .await .unwrap(); - assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); - let saved = std::fs::read_to_string(&config_path).unwrap(); - assert_eq!(saved, original); - assert_eq!(crate::config::config_revision(&saved), base_revision); - assert!(config_temporary_files(&config_path).is_empty()); - assert!(!service_control.restart_pending()); + assert_eq!(response.status(), StatusCode::OK); + assert!(sessions.is_valid(&token).await.unwrap()); + assert_eq!( + std::fs::read_to_string(&config_path).unwrap(), + candidate_toml + ); + expect_restart_completed(&mut restarts).await; let _ = std::fs::remove_file(config_path); } @@ -969,7 +1067,7 @@ mod route_tests { } #[tokio::test] - async fn password_change_commit_failure_invalidates_sessions_and_returns_internal_error() { + async fn password_change_commit_failure_preserves_sessions_and_returns_internal_error() { let mut state = test_state(); let (config_path, base_revision) = write_config_file(&state.config, "password-commit-failure"); @@ -1009,7 +1107,7 @@ mod route_tests { ) .unwrap(); assert_eq!(body["error"]["code"], "internal_error"); - assert!(!sessions.is_valid(&token).await.unwrap()); + assert!(sessions.is_valid(&token).await.unwrap()); let saved = std::fs::read_to_string(&config_path).unwrap(); assert_eq!(saved, original); assert_eq!(crate::config::config_revision(&saved), base_revision); diff --git a/src/api/service.rs b/src/api/service.rs index 5882a22..5d770ed 100644 --- a/src/api/service.rs +++ b/src/api/service.rs @@ -1,5 +1,5 @@ use std::process::Command; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -20,16 +20,28 @@ struct StatusResponse { api_bind: String, api_port: u16, database_path: String, + restart_status: RestartStatus, } pub trait ServiceRestarter: Send + Sync { - fn restart(&self); + fn restart(&self) -> anyhow::Result<()>; +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[repr(u8)] +#[serde(rename_all = "snake_case")] +pub enum RestartStatus { + Idle, + Scheduled, + CommandCompleted, + CommandFailed, } #[derive(Clone)] pub struct ServiceControl { restarter: Arc, restart_pending: Arc, + restart_status: Arc, } impl ServiceControl { @@ -37,6 +49,7 @@ impl ServiceControl { Self { restarter: Arc::new(restarter), restart_pending: Arc::new(AtomicBool::new(false)), + restart_status: Arc::new(AtomicU8::new(RestartStatus::Idle as u8)), } } @@ -51,16 +64,38 @@ impl ServiceControl { let restarter = self.restarter.clone(); let restart_pending = self.restart_pending.clone(); + let restart_status = self.restart_status.clone(); + restart_status.store(RestartStatus::Scheduled as u8, Ordering::Release); tokio::spawn(async move { tokio::time::sleep(Duration::from_millis(500)).await; - if let Err(error) = tokio::task::spawn_blocking(move || restarter.restart()).await { - log::warn!("service restart task failed: {}", error); - } + let status = match tokio::task::spawn_blocking(move || restarter.restart()).await { + Ok(Ok(())) => RestartStatus::CommandCompleted, + Ok(Err(error)) => { + log::warn!("service restart command failed: {}", error); + RestartStatus::CommandFailed + } + Err(error) => { + log::warn!("service restart task failed: {}", error); + RestartStatus::CommandFailed + } + }; + restart_status.store(status as u8, Ordering::Release); restart_pending.store(false, Ordering::Release); }); true } + pub fn restart_status(&self) -> RestartStatus { + match self.restart_status.load(Ordering::Acquire) { + value if value == RestartStatus::Scheduled as u8 => RestartStatus::Scheduled, + value if value == RestartStatus::CommandCompleted as u8 => { + RestartStatus::CommandCompleted + } + value if value == RestartStatus::CommandFailed as u8 => RestartStatus::CommandFailed, + _ => RestartStatus::Idle, + } + } + #[cfg(test)] pub(crate) fn restart_pending(&self) -> bool { self.restart_pending.load(Ordering::Acquire) @@ -76,26 +111,20 @@ impl Default for ServiceControl { struct SystemServiceRestarter; impl ServiceRestarter for SystemServiceRestarter { - fn restart(&self) { + fn restart(&self) -> anyhow::Result<()> { let initd = "/etc/init.d/sms-relayed"; - let result = if std::path::Path::new(initd).exists() { + let status = if std::path::Path::new(initd).exists() { Command::new(initd).arg("restart").status() } else { Command::new("systemctl") .args(["restart", "sms-relayed"]) .status() - }; - match result { - Ok(status) if status.success() => { - log::info!("service restart command completed successfully"); - } - Ok(status) => { - log::warn!("service restart command exited with status {}", status); - } - Err(error) => { - log::warn!("failed to run service restart command: {}", error); - } + }?; + if !status.success() { + anyhow::bail!("service restart command exited with status {status}"); } + log::info!("service restart command completed successfully"); + Ok(()) } } @@ -120,6 +149,7 @@ async fn status(State(state): State) -> ApiResult api_bind: state.config.api.bind.clone(), api_port: state.config.api.port, database_path: state.config.api.database_path.clone(), + restart_status: state.service_control.restart_status(), })) } diff --git a/src/dbus/inbound.rs b/src/dbus/inbound.rs index f9a750e..d3fd485 100644 --- a/src/dbus/inbound.rs +++ b/src/dbus/inbound.rs @@ -5,7 +5,7 @@ use std::time::Duration; use anyhow::Result; use futures_util::{future::BoxFuture, StreamExt}; use zbus::names::OwnedUniqueName; -use zbus::zvariant::{OwnedObjectPath, OwnedValue}; +use zbus::zvariant::{OwnedObjectPath, OwnedValue, Value}; use zbus::{Connection, Message, MessageStream}; use super::{ @@ -15,6 +15,7 @@ use super::{ const DBUS_METHOD_TIMEOUT: Duration = Duration::from_secs(10); const DBUS_PROPERTIES_TIMEOUT: Duration = Duration::from_secs(5); +const MM_SMS_STATE_RECEIVED: u32 = 3; #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct InboundSmsProperties { @@ -95,15 +96,18 @@ impl SystemInboundSource { add_match_rule(&connection, &removed_rule).await?; let reader = Arc::new(ZbusSmsPropertiesReader { - connection, + connection: connection.clone(), owner: owner.clone(), }); + let initial_sms = + list_received_sms(&connection, &owner, modem_path, reader.clone()).await?; Ok(InboundSubscription::new( modem_path.to_string(), owner.to_string(), Box::new(ZbusSessionBackend { stream }), reader, - )) + ) + .with_initial_sms(initial_sms)) } } @@ -112,6 +116,7 @@ pub(crate) struct InboundSubscription { owner: String, backend: Box, reader: Arc, + initial_sms: Vec, terminal_error: Option<&'static str>, } @@ -127,10 +132,20 @@ impl InboundSubscription { owner, backend, reader, + initial_sms: Vec::new(), terminal_error: None, } } + fn with_initial_sms(mut self, initial_sms: Vec) -> Self { + self.initial_sms = initial_sms; + self + } + + pub(crate) fn take_initial_sms(&mut self) -> Vec { + std::mem::take(&mut self.initial_sms) + } + pub(crate) async fn next(&mut self) -> Result { if let Some(error) = self.terminal_error { return Err(anyhow::anyhow!(error)); @@ -179,6 +194,63 @@ impl InboundSubscription { } } +async fn list_received_sms( + connection: &Connection, + owner: &OwnedUniqueName, + modem_path: &str, + reader: Arc, +) -> Result> { + let list_call = connection.call_method( + Some(owner), + modem_path, + Some(MM_MESSAGING_INTERFACE), + "List", + &(), + ); + let list_reply = tokio::time::timeout(DBUS_METHOD_TIMEOUT, list_call) + .await + .map_err(|_| anyhow::anyhow!("dbus SMS list timeout"))??; + let paths: Vec = list_reply.body().deserialize()?; + let mut received = Vec::with_capacity(paths.len()); + for path in paths { + let sms_path = path.to_string(); + if sms_is_received(connection, owner, &sms_path).await? { + received.push(InboundSms { + path: sms_path, + reader: reader.clone(), + }); + } + } + Ok(received) +} + +async fn sms_is_received( + connection: &Connection, + owner: &OwnedUniqueName, + sms_path: &str, +) -> Result { + let call = connection.call_method( + Some(owner), + sms_path, + Some(DBUS_PROPERTIES_INTERFACE), + "GetAll", + &(MM_SMS_INTERFACE,), + ); + let reply = tokio::time::timeout(DBUS_PROPERTIES_TIMEOUT, call) + .await + .map_err(|_| anyhow::anyhow!("dbus SMS snapshot state timeout"))??; + let properties: HashMap = reply.body().deserialize()?; + let state = properties + .get("State") + .ok_or_else(|| anyhow::anyhow!("dbus SMS snapshot has no State property"))?; + let Value::U32(state) = (**state).clone() else { + return Err(anyhow::anyhow!( + "dbus SMS snapshot State property has unexpected type" + )); + }; + Ok(state == MM_SMS_STATE_RECEIVED) +} + trait SessionBackend: Send { fn next_signal<'a>(&'a mut self) -> BoxFuture<'a, Result>>; } diff --git a/src/inbound.rs b/src/inbound.rs index 196498b..73f087d 100644 --- a/src/inbound.rs +++ b/src/inbound.rs @@ -1,7 +1,7 @@ use std::sync::Arc; use std::time::Duration; -use anyhow::Result; +use anyhow::{Context, Result}; use futures_util::future::BoxFuture; use log::{error, info, warn}; use tokio::sync::Semaphore; @@ -20,6 +20,15 @@ const INITIAL_RECONNECT_DELAY: Duration = Duration::from_secs(5); const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(60); const BODY_POLL_INTERVAL: Duration = Duration::from_millis(100); const MAX_BODY_POLLS: usize = 600; +const MAX_PROPERTY_READ_ATTEMPTS: usize = 5; +#[cfg(not(test))] +const INITIAL_PROPERTY_READ_RETRY_DELAY: Duration = Duration::from_millis(100); +#[cfg(test)] +const INITIAL_PROPERTY_READ_RETRY_DELAY: Duration = Duration::from_millis(1); +#[cfg(not(test))] +const MAX_PROPERTY_READ_RETRY_DELAY: Duration = Duration::from_secs(2); +#[cfg(test)] +const MAX_PROPERTY_READ_RETRY_DELAY: Duration = Duration::from_millis(4); const INITIAL_PERSISTENCE_RETRY_DELAY: Duration = Duration::from_millis(100); const MAX_PERSISTENCE_RETRY_DELAY: Duration = Duration::from_secs(30); const LEGACY_SINGLE_MODEM_FINGERPRINT_SEED: &str = "sms-relayed-single-modem"; @@ -165,7 +174,7 @@ impl InboundWorker { } let Some(path) = current_path.clone() else { warn!("no runtime modem path available; retrying resolution"); - tokio::time::sleep(delay).await; + wait_for_reconnect(delay, &mut children).await; delay = next_reconnect_delay(delay); continue; }; @@ -205,12 +214,16 @@ impl InboundWorker { } info!("reconnecting in {}s...", delay.as_secs_f64()); - tokio::time::sleep(delay).await; + wait_for_reconnect(delay, &mut children).await; delay = next_reconnect_delay(delay); } } - async fn run_subscription(&self, actual_path: &str, children: &mut JoinSet<()>) -> Result<()> { + async fn run_subscription( + &self, + actual_path: &str, + children: &mut JoinSet>, + ) -> Result<()> { let mut subscription = self.source.subscribe(actual_path).await?; let mut identity_refresh = tokio::time::interval(RUNTIME_IDENTITY_REFRESH_INTERVAL); identity_refresh.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); @@ -218,6 +231,10 @@ impl InboundWorker { info!("SMS monitor ready on {}", actual_path); + for sms in subscription.take_initial_sms() { + self.spawn_incoming_sms(sms, children).await?; + } + loop { let sms = tokio::select! { _ = identity_refresh.tick() => { @@ -232,29 +249,33 @@ impl InboundWorker { } sms = subscription.next_sms() => sms?, }; - info!("SmsPath:\n{}", sms.path()); - - let permit = self - .inbound_limit - .clone() - .acquire_owned() - .await - .map_err(|_| anyhow::anyhow!("inbound task limiter closed"))?; - let storage_filters = self.settings.ignored_storage.clone(); - let profile_keys = self.settings.profile_keys.clone(); - let messaging = self.messaging.clone(); - children.spawn(async move { - let _permit = permit; - if process_incoming_sms(sms, &storage_filters, messaging, profile_keys) - .await - .is_err() - { - report_child_failure(); - } - }); + self.spawn_incoming_sms(sms, children).await?; } } + async fn spawn_incoming_sms( + &self, + sms: Box, + children: &mut JoinSet>, + ) -> Result<()> { + info!("SmsPath:\n{}", sms.path()); + + let permit = self + .inbound_limit + .clone() + .acquire_owned() + .await + .map_err(|_| anyhow::anyhow!("inbound task limiter closed"))?; + let storage_filters = self.settings.ignored_storage.clone(); + let profile_keys = self.settings.profile_keys.clone(); + let messaging = self.messaging.clone(); + children.spawn(async move { + let _permit = permit; + process_incoming_sms(sms, &storage_filters, messaging, profile_keys).await + }); + Ok(()) + } + async fn observe_runtime_identity(&self, actual_path: &str) -> Result<()> { retry_pending_identity_mismatches(&self.modem_service, &self.store).await?; let action_path = self.modem_service.verified_path(); @@ -328,17 +349,37 @@ impl InboundWorker { } } -fn report_child_result(joined: Option>) { - if matches!(joined, Some(Err(_))) { - report_child_failure(); +fn report_child_result(joined: Option, tokio::task::JoinError>>) { + match joined { + Some(Ok(Ok(()))) | None => {} + Some(Ok(Err(error))) => { + error!("incoming SMS processing task failed: {error:#}"); + report_child_failure(); + } + Some(Err(error)) => { + error!("incoming SMS processing task failed to join: {error}"); + report_child_failure(); + } } } fn report_child_failure() { - error!("incoming SMS processing task failed"); crate::monitoring::capture_failure("dbus", "dbus.inbound_processing_failed"); } +async fn wait_for_reconnect(delay: Duration, children: &mut JoinSet>) { + let sleep = tokio::time::sleep(delay); + tokio::pin!(sleep); + loop { + tokio::select! { + _ = &mut sleep => return, + joined = children.join_next(), if !children.is_empty() => { + report_child_result(joined); + } + } + } +} + fn publish_resolved_path(modem_service: &ModemService, resolved: ModemTargets) -> Option { let path = resolved.runtime_path().map(ToString::to_string); modem_service.set_modem_targets(resolved); @@ -389,6 +430,42 @@ fn next_persistence_retry_delay(delay: Duration) -> Duration { (delay * 2).min(MAX_PERSISTENCE_RETRY_DELAY) } +fn next_property_read_retry_delay(delay: Duration) -> Duration { + (delay * 2).min(MAX_PROPERTY_READ_RETRY_DELAY) +} + +async fn read_sms_properties_with_retry( + sms: &dyn InboundSmsAdapter, +) -> Result { + let mut delay = INITIAL_PROPERTY_READ_RETRY_DELAY; + for attempt in 1..=MAX_PROPERTY_READ_ATTEMPTS { + match sms.properties().await { + Ok(properties) => return Ok(properties), + Err(error) if attempt == MAX_PROPERTY_READ_ATTEMPTS => { + return Err(error).with_context(|| { + format!( + "failed to read SMS properties for {} after {} attempts", + sms.path(), + MAX_PROPERTY_READ_ATTEMPTS + ) + }); + } + Err(error) => { + warn!( + "read SMS properties failed for {} on attempt {}/{}; retrying: {}", + sms.path(), + attempt, + MAX_PROPERTY_READ_ATTEMPTS, + error + ); + tokio::time::sleep(delay).await; + delay = next_property_read_retry_delay(delay); + } + } + } + unreachable!("property read loop returns on success or final failure") +} + async fn process_incoming_sms( sms: Box, storage_filters: &[StorageType], @@ -397,7 +474,7 @@ async fn process_incoming_sms( ) -> Result<()> { let mut retries = 0; loop { - let properties = sms.properties().await?; + let properties = read_sms_properties_with_retry(sms.as_ref()).await?; if should_ignore_storage(properties.storage, storage_filters) { warn!("已过滤不转发"); @@ -641,6 +718,8 @@ trait InboundSourceAdapter: Send + Sync { } trait InboundSubscriptionAdapter: Send { + fn take_initial_sms(&mut self) -> Vec>; + fn next_sms<'a>(&'a mut self) -> BoxFuture<'a, Result>>; } @@ -668,6 +747,14 @@ impl InboundSourceAdapter for SystemSourceAdapter { struct SystemSubscriptionAdapter(InboundSubscription); impl InboundSubscriptionAdapter for SystemSubscriptionAdapter { + fn take_initial_sms(&mut self) -> Vec> { + self.0 + .take_initial_sms() + .into_iter() + .map(|sms| Box::new(SystemSmsAdapter(sms)) as Box) + .collect() + } + fn next_sms<'a>(&'a mut self) -> BoxFuture<'a, Result>> { Box::pin(async move { let InboundEvent::Added(sms) = self.0.next().await?; @@ -779,6 +866,10 @@ mod tests { } impl InboundSubscriptionAdapter for ScriptedSubscription { + fn take_initial_sms(&mut self) -> Vec> { + Vec::new() + } + fn next_sms<'a>(&'a mut self) -> BoxFuture<'a, Result>> { self.reads.fetch_add(1, Ordering::SeqCst); let message = self.messages.pop_front(); @@ -798,7 +889,7 @@ mod tests { } impl ScriptedSource { - fn new(subscription: ScriptedSubscription) -> Self { + fn new(subscription: impl InboundSubscriptionAdapter + 'static) -> Self { Self { subscription: Mutex::new(Some(Box::new(subscription))), subscribed: Arc::new(Notify::new()), @@ -1084,6 +1175,141 @@ mod tests { .expect("counter reached expected value"); } + async fn wait_for_messages(query: &Messaging) -> Vec { + tokio::time::timeout(Duration::from_secs(1), async { + loop { + let messages = query + .list(crate::message::MessageFilter::default()) + .await + .unwrap(); + if !messages.is_empty() { + break messages; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("inbound SMS reaches the messaging query seam") + } + + #[tokio::test] + async fn reconnect_delay_drains_completed_processing_tasks() { + let mut children = JoinSet::new(); + children.spawn(async { Err(anyhow::anyhow!("last property source")) }); + + wait_for_reconnect(Duration::from_millis(10), &mut children).await; + + assert!(children.is_empty()); + } + + struct SnapshotSubscription { + initial_messages: Vec>, + live_messages: VecDeque>, + live_reads: Arc, + } + + impl InboundSubscriptionAdapter for SnapshotSubscription { + fn take_initial_sms(&mut self) -> Vec> { + std::mem::take(&mut self.initial_messages) + } + + fn next_sms<'a>(&'a mut self) -> BoxFuture<'a, Result>> { + self.live_reads.fetch_add(1, Ordering::SeqCst); + let message = self.live_messages.pop_front(); + Box::pin(async move { + match message { + Some(message) => Ok(message), + None => std::future::pending().await, + } + }) + } + } + + #[tokio::test] + async fn startup_snapshot_is_persisted_without_a_live_added_signal() { + let source = Arc::new(ScriptedSource::new(SnapshotSubscription { + initial_messages: vec![Box::new(ScriptedSms::new( + SMS_PATH, + vec![PropertyAction::Return(Ok(properties( + "arrived while offline", + StorageType::Me as u32, + )))], + ))], + live_messages: VecDeque::new(), + live_reads: Arc::new(AtomicUsize::new(0)), + })); + let store = Store::open_in_memory().unwrap(); + store + .set_modem_fingerprint("snapshot-fingerprint".to_string()) + .await + .unwrap(); + let query = messaging(store.clone()); + let worker = worker(store, source, settings(Vec::new(), Vec::new())); + + let run = tokio::spawn(async move { + let mut children = JoinSet::new(); + worker.run_subscription(MODEM_PATH, &mut children).await + }); + + let messages = wait_for_messages(&query).await; + + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].body, "arrived while offline"); + run.abort(); + let _ = run.await; + } + + #[tokio::test] + async fn startup_snapshot_overlapping_a_live_signal_is_persisted_once() { + let live_reads = Arc::new(AtomicUsize::new(0)); + let duplicate_properties = || { + PropertyAction::Return(Ok(properties( + "same SMS from snapshot and signal", + StorageType::Me as u32, + ))) + }; + let source = Arc::new(ScriptedSource::new(SnapshotSubscription { + initial_messages: vec![Box::new(ScriptedSms::new( + SMS_PATH, + vec![duplicate_properties()], + ))], + live_messages: vec![ + Box::new(ScriptedSms::new(SMS_PATH, vec![duplicate_properties()])) + as Box, + ] + .into(), + live_reads: live_reads.clone(), + })); + let store = Store::open_in_memory().unwrap(); + store + .set_modem_fingerprint("snapshot-overlap-fingerprint".to_string()) + .await + .unwrap(); + let query = messaging(store.clone()); + let worker = worker(store, source, settings(Vec::new(), Vec::new())); + + let run = tokio::spawn(async move { + let mut children = JoinSet::new(); + worker.run_subscription(MODEM_PATH, &mut children).await + }); + + wait_for_count(&live_reads, 1).await; + let messages = wait_for_messages(&query).await; + tokio::time::sleep(Duration::from_millis(20)).await; + + assert_eq!(messages.len(), 1); + assert_eq!( + query + .list(crate::message::MessageFilter::default()) + .await + .unwrap() + .len(), + 1 + ); + run.abort(); + let _ = run.await; + } + #[tokio::test] async fn seventeenth_sms_backpressures_subscription_before_spawn() { let started = Arc::new(AtomicUsize::new(0)); @@ -1215,6 +1441,80 @@ mod tests { ); } + #[tokio::test] + async fn transient_property_failure_retries_then_reaches_messaging_query() { + let store = Store::open_in_memory().unwrap(); + store + .set_modem_fingerprint("property-retry-fingerprint".to_string()) + .await + .unwrap(); + let query = messaging(store); + let sms = ScriptedSms::new( + SMS_PATH, + vec![ + PropertyAction::Return(Err(anyhow::anyhow!("temporary property timeout"))), + PropertyAction::Return(Ok(properties( + "available after retry", + StorageType::Me as u32, + ))), + ], + ); + let calls = sms.calls.clone(); + + process_incoming_sms(Box::new(sms), &[], query.clone(), Vec::new()) + .await + .unwrap(); + + let messages = query + .list(crate::message::MessageFilter::default()) + .await + .unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), 2); + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].body, "available after retry"); + } + + #[tokio::test] + async fn terminal_property_failure_preserves_path_attempts_and_last_source() { + let store = Store::open_in_memory().unwrap(); + store + .set_modem_fingerprint("terminal-property-failure".to_string()) + .await + .unwrap(); + let expected_attempts = 5; + let mut actions = (1..expected_attempts) + .map(|attempt| { + PropertyAction::Return(Err(anyhow::anyhow!("earlier property failure {attempt}"))) + }) + .collect::>(); + actions.push(PropertyAction::Return(Err(anyhow::anyhow!( + "last property source" + )))); + actions.push(PropertyAction::Wait { + started: Arc::new(AtomicUsize::new(0)), + dropped: Arc::new(AtomicUsize::new(0)), + }); + let sms = ScriptedSms::new(SMS_PATH, actions); + let calls = sms.calls.clone(); + + let result = tokio::time::timeout( + Duration::from_secs(1), + process_incoming_sms(Box::new(sms), &[], messaging(store), Vec::new()), + ) + .await + .expect("property retry budget must be bounded"); + let error = result.unwrap_err(); + let error_chain = format!("{error:#}"); + + assert_eq!(calls.load(Ordering::SeqCst), expected_attempts); + assert!(error_chain.contains(SMS_PATH), "{error_chain}"); + assert!(error_chain.contains("after 5 attempts"), "{error_chain}"); + assert!( + error_chain.contains("last property source"), + "{error_chain}" + ); + } + #[tokio::test] async fn body_is_polled_until_available_then_persisted() { let store = Store::open_in_memory().unwrap(); diff --git a/src/persistence/auth.rs b/src/persistence/auth.rs index ee3ec07..ec918e1 100644 --- a/src/persistence/auth.rs +++ b/src/persistence/auth.rs @@ -47,10 +47,6 @@ impl Store { .await } - pub async fn delete_all_auth_sessions(&self) -> Result<()> { - self.run(|sqlite| sqlite.delete_all_auth_sessions()).await - } - #[cfg(test)] pub async fn expire_auth_session(&self, token_hash: Vec) -> Result<()> { self.run(move |sqlite| sqlite.expire_auth_session(&token_hash)) diff --git a/src/storage/auth_sessions.rs b/src/storage/auth_sessions.rs index aaa9dfc..a7b0e8e 100644 --- a/src/storage/auth_sessions.rs +++ b/src/storage/auth_sessions.rs @@ -109,12 +109,6 @@ impl MessageStore { Ok(()) } - pub fn delete_all_auth_sessions(&self) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute("DELETE FROM auth_sessions", [])?; - Ok(()) - } - #[cfg(test)] pub fn expire_auth_session(&self, token_hash: &[u8]) -> Result<()> { let conn = self.conn.lock().unwrap();