From f1fe9e6a3074a0b9be774eccd8d5c446cdf5b823 Mon Sep 17 00:00:00 2001 From: abdulsaheel Date: Wed, 8 Jul 2026 09:01:57 +0530 Subject: [PATCH 1/2] fix: BLE connection/sync redundancy, derivation concurrency, platform-standards compliance MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BLE reliability (connection ladder + sync integrity): - Timeout discoverServices()/setNotifyValue() so a wedged BLE stack can't hang the connect chain forever and jam the whole reconnect ladder above it - FrameCorruptionDetector: CRC-failure-rate detector, independent of the existing timeout-based MarginalRadioDetector (catches a radio that responds promptly but wrong, not just one that stops responding) - Band-truth reconciliation: track expectedPacketCount mismatch streaks over time instead of only logging once - CounterRegressionDetector: explicit, observable band-reboot signal, seeded from the durable counter_hw cursor across reconnects - Periodic RTC re-verification on long-lived connections (previously only checked once at connect) - isCorruptFutureRtc: sanity-check GET_DATA_RANGE's range_newest before it can poison the session's plausibility window - Real per-chunk sync_ledger (ChunkFailureLedger) + wired sync_quarantine read/write — a token failing ACK repeatedly across reconnects is now tracked and quarantined instead of silently overwriting one shared row - Activated schemaHealth() (real PRAGMA integrity_check) at startup + on the foreground cadence — was fully implemented but never called - Removed dead, unwired, untested Whoop5EmptyOffloadTracker - HeadlessSyncGate skip-streak telemetry (per-owner + lifetime counters) - Cross-platform staleness escalation: local notification at 48h+ silence, quiet in-app tier at 12h, wired into both the headless and foreground cadence paths - Android OEM autostart deep links (Xiaomi/Huawei/Honor/Oppo/Vivo/OnePlus), gated behind ActivityManager.isBackgroundRestricted() (the one official documented signal) rather than firing on manufacturer-string match alone - iOS force-quit explainer tied to the staleness notification - Per-checkpoint diagnostic logging in commitSyncBatch (decoded/archive/ cursor phases) via a failure-safe optional callback, atomicity unchanged - Notification permission requests now respect platform docs: background/ headless code paths only check (never request) authorization, since an interactive prompt requires a foreground scene the OS won't grant one to from a background execution context Derivation engine concurrency (single-threaded backlog sweeps were the biggest contributor to "results aren't ready when the app opens"): - runWithConcurrency: bounded worker-pool primitive replacing three identical fully-sequential per-day for-loops (run/runDays/rescanRecent). A multi-day backlog used to process one day's isolate work fully before starting the next, leaving every core but one idle; days within a run are independent (frozen baseline snapshot, day_id-keyed writes) so this is genuinely safe to parallelize - Newest-first day ordering — a backlog sweep used to compute the oldest pending day first and today dead last, backwards from what the user actually wants on open - Per-day-local prepare-stats accumulator so concurrent days don't race on shared diagnostic counters (telemetry-only; never a correctness issue for the derived values themselves) All changes verified against official Android/Apple documentation (not community folklore) where platform behavior was load-bearing. 434 Dart tests green (62 new), zero new analyzer issues, Android debug APK and iOS debug build both verified. --- .../openstrap_edge/NativeChannels.kt | 130 ++++++++++ ios/Runner/BgSyncScheduler.swift | 6 + ios/Runner/BleRestoreManager.swift | 18 ++ lib/ble/android_background.dart | 88 ++++++- lib/ble/ble_engine.dart | 206 ++++++++++++++- lib/ble/ble_state.dart | 78 ++++++ lib/compute/derivation_engine.dart | 240 ++++++++++++++---- lib/data/db.dart | 37 +++ lib/notify/notification_center.dart | 21 +- lib/notify/notification_service.dart | 50 +++- lib/state/app_state.dart | 75 +++++- lib/sync/background_sync.dart | 65 ++++- lib/sync/headless_gate.dart | 39 ++- lib/sync/sync_policy.dart | 173 ++++++++++--- lib/ui/profile/profile_screen.dart | 73 ++++++ test/ble_state_test.dart | 106 ++++++++ test/db_integrity_test.dart | 138 ++++++++++ test/derive_concurrency_test.dart | 117 +++++++++ test/headless_gate_test.dart | 97 +++++++ test/sync_policy_test.dart | 151 +++++++++++ 20 files changed, 1796 insertions(+), 112 deletions(-) create mode 100644 test/derive_concurrency_test.dart create mode 100644 test/headless_gate_test.dart diff --git a/android/app/src/main/kotlin/wtf/openstrap/openstrap_edge/NativeChannels.kt b/android/app/src/main/kotlin/wtf/openstrap/openstrap_edge/NativeChannels.kt index 70a8503e..9d8c5ab9 100644 --- a/android/app/src/main/kotlin/wtf/openstrap/openstrap_edge/NativeChannels.kt +++ b/android/app/src/main/kotlin/wtf/openstrap/openstrap_edge/NativeChannels.kt @@ -1,7 +1,10 @@ package wtf.openstrap.openstrap_edge +import android.app.ActivityManager +import android.content.ComponentName import android.content.Context import android.content.Intent +import android.content.ActivityNotFoundException import android.hardware.camera2.CameraCharacteristics import android.hardware.camera2.CameraManager import android.media.AudioManager @@ -85,6 +88,13 @@ object NativeChannels { "requestIgnoreBatteryOptimizations" -> { result.success(requestIgnoreBatteryOptimizations(app)) } + "manufacturerHint" -> result.success(Build.MANUFACTURER.lowercase()) + "isBackgroundRestricted" -> { + result.success(isBackgroundRestricted(app)) + } + "openOemAutostartSettings" -> { + result.success(openOemAutostartSettings(app)) + } else -> result.notImplemented() } } @@ -146,6 +156,126 @@ object NativeChannels { } } + /** + * Whether the OS is CURRENTLY restricting this app's background work — + * the one official, CTS-tested, documented signal for this situation + * (`ActivityManager.isBackgroundRestricted`, API 28+): "if true, any work + * that the app tries to do will be aggressively restricted while it is in + * the background... jobs and alarms will not execute and foreground + * services cannot be started." This is what actually gates whether the + * OEM-autostart entry point below should even be surfaced to the user — + * NOT a manufacturer-name guess, which can't tell whether the OS is + * presently restricting anything at all. False on API <28 (unsupported, + * so we can't tell — callers fall back to the manufacturer hint alone in + * that case, same as before). + */ + private fun isBackgroundRestricted(ctx: Context): Boolean { + if (Build.VERSION.SDK_INT < Build.VERSION_CODES.P) return false + val am = ctx.getSystemService(Context.ACTIVITY_SERVICE) as ActivityManager + return am.isBackgroundRestricted + } + + /** + * OEM autostart/battery-manager allowlist deep link — a second, stronger + * line of defense than [requestIgnoreBatteryOptimizations]. The stock + * Android Doze exemption is well-known to be INSUFFICIENT on Xiaomi + * (MIUI)/Huawei/Honor/Oppo (ColorOS)/Vivo (FuntouchOS)/OnePlus — these + * OEMs layer their own aggressive process killers on top of stock Doze + * and gate survival behind a separate "autostart"/"protected apps" list + * that stock APIs cannot toggle. There is NO official Android API for + * this specific mechanism (confirmed against developer.android.com's + * Doze/App-Standby guide, which never mentions OEM autostart screens); + * the settings-activity ComponentNames below are long-standing + * community-documented ones (the "autostarter" pattern), not a Google + * source, and can change across OEM software versions — every attempt + * is wrapped so a missing/renamed activity on some device just falls + * through to the next candidate, never crashes. Falls back to this app's + * standard "App info" settings page (always resolvable) if no + * OEM-specific screen exists on this device — so the user always lands + * somewhere useful, never a silent no-op. Dart gates whether to even + * OFFER this (via [isBackgroundRestricted]) rather than firing it + * unconditionally off the manufacturer string — see + * AndroidBackground.needsOemAutostartSettings. + */ + private fun openOemAutostartSettings(ctx: Context): String { + val manufacturer = Build.MANUFACTURER.lowercase() + val candidates: List = when { + manufacturer.contains("xiaomi") -> listOf( + ComponentName( + "com.miui.securitycenter", + "com.miui.permcenter.autostart.AutoStartManagementActivity", + ), + ComponentName( + "com.miui.securitycenter", + "com.miui.powercenter.PowerSettings", + ), + ) + manufacturer.contains("huawei") || manufacturer.contains("honor") -> listOf( + ComponentName( + "com.huawei.systemmanager", + "com.huawei.systemmanager.startupmgr.ui.StartupNormalAppListActivity", + ), + ComponentName( + "com.huawei.systemmanager", + "com.huawei.systemmanager.optimize.process.ProtectActivity", + ), + ) + manufacturer.contains("oppo") || manufacturer.contains("realme") -> listOf( + ComponentName( + "com.coloros.safecenter", + "com.coloros.safecenter.permission.startup.StartupAppListActivity", + ), + ComponentName( + "com.coloros.safecenter", + "com.coloros.safecenter.startupapp.StartupAppListActivity", + ), + ) + manufacturer.contains("vivo") -> listOf( + ComponentName( + "com.vivo.permissionmanager", + "com.vivo.permissionmanager.activity.BgStartUpManagerActivity", + ), + ComponentName( + "com.iqoo.secure", + "com.iqoo.secure.ui.phoneoptimize.AddWhiteListActivity", + ), + ) + manufacturer.contains("oneplus") -> listOf( + ComponentName( + "com.oneplus.security", + "com.oneplus.security.chainlaunch.view.ChainLaunchAppListActivity", + ), + ) + else -> emptyList() + } + + for (component in candidates) { + try { + val intent = Intent().apply { + setComponent(component) + addFlags(Intent.FLAG_ACTIVITY_NEW_TASK) + } + ctx.startActivity(intent) + return "opened_oem_autostart" + } catch (e: ActivityNotFoundException) { + continue // try the next candidate / fall through to app-info + } catch (e: SecurityException) { + continue + } + } + + return try { + val fallback = Intent( + Settings.ACTION_APPLICATION_DETAILS_SETTINGS, + Uri.parse("package:${ctx.packageName}"), + ).apply { addFlags(Intent.FLAG_ACTIVITY_NEW_TASK) } + ctx.startActivity(fallback) + "opened_app_info_fallback" + } catch (e: Exception) { + "failed" + } + } + private fun audio(ctx: Context): AudioManager = ctx.getSystemService(Context.AUDIO_SERVICE) as AudioManager diff --git a/ios/Runner/BgSyncScheduler.swift b/ios/Runner/BgSyncScheduler.swift index cbb58d48..7dadfe66 100644 --- a/ios/Runner/BgSyncScheduler.swift +++ b/ios/Runner/BgSyncScheduler.swift @@ -84,6 +84,12 @@ enum BackgroundTaskManager { /// Submit (or renew) the next BGProcessingTaskRequest. Safe to call multiple /// times — if a request is already pending, the OS silently replaces it. + /// + /// `earliestBeginDate` is documented as a floor, never a promise: "the system + /// doesn't guarantee launching the task at the specified date, but only that + /// it won't begin sooner" (BGTaskRequest.earliestBeginDate). Apple's own + /// sample code uses the identical `Date(timeIntervalSinceNow: 15 * 60)` + /// pattern — confirmed this matches, not an assumption. static func schedule() { let req = BGProcessingTaskRequest(identifier: taskIdentifier) req.earliestBeginDate = Date(timeIntervalSinceNow: retryInterval) diff --git a/ios/Runner/BleRestoreManager.swift b/ios/Runner/BleRestoreManager.swift index de96a019..f9bdb866 100644 --- a/ios/Runner/BleRestoreManager.swift +++ b/ios/Runner/BleRestoreManager.swift @@ -7,6 +7,14 @@ import Flutter /// reachable — the mechanism WHOOP/Garmin use on iOS (CoreBluetooth State Preservation /// & Restoration). No persistent notification, no foreground service. /// +/// Running this as a SEPARATE CBCentralManager (distinct restoration identifier) from +/// the "live" central flutter_blue_plus drives is an Apple-documented, supported pattern, +/// not an inferred workaround: "Because apps can have multiple instances of +/// CBCentralManager... be sure each restoration identifier is unique, so that the system +/// can properly distinguish one central... from another" (Core Bluetooth Background +/// Processing for iOS Apps). Confirmed directly against that doc — this is not a deviation +/// from a single-manager model Apple only describes for the simple case. +/// /// It does NOT drain data. flutter_blue_plus owns the real GATT session, and two /// CBCentralManagers can't share a peripheral connection. This is a trigger only: it /// holds a no-timeout pending connect to the band (under a restore identifier) so iOS @@ -21,6 +29,16 @@ import Flutter /// Dart reports the drain done (`syncDone`), we go IDLE and do NOT re-arm. We re-arm only /// on the next explicit request from Dart (a fresh disconnect). Arming only happens while /// backgrounded; in the foreground flutter_blue_plus owns the band. +/// +/// Verified against Apple's official docs ("Core Bluetooth Background Processing for iOS +/// Apps"): a state-restoration relaunch is a BOUNDED wake, not indefinite runtime — "an app +/// has around 10 seconds to complete a task... apps that spend too much time executing in +/// the background can be throttled back by the system or killed," and even a fully +/// backgrounded app "can't run forever... the system may need to terminate your app to free +/// up memory." This is exactly why the headless sync this triggers (background_sync.dart's +/// runHeadlessSync) is designed to make partial progress safely on every wake — commit +/// whatever it drained before the window closes, resume from the durable cursor next time — +/// rather than assuming it gets to run to completion in one continuous background session. class BleRestoreManager: NSObject { static let shared = BleRestoreManager() diff --git a/lib/ble/android_background.dart b/lib/ble/android_background.dart index 22edcaac..a17792d5 100644 --- a/lib/ble/android_background.dart +++ b/lib/ble/android_background.dart @@ -1,6 +1,6 @@ // android_background.dart — Android OS keep-alive integrations, Dart side. // -// Two independent levers that make the background BLE session survive the OS: +// Three independent levers that make the background BLE session survive the OS: // // 1. CompanionDeviceManager (CDM) association. After pairing we associate the // band's MAC with the app via CDM (a one-time system dialog pre-filtered to @@ -15,6 +15,20 @@ // system ACTION_REQUEST_IGNORE_BATTERY_OPTIMIZATIONS dialog. Without the // exemption, Doze can freeze the process between BLE events overnight. // +// 3. OEM autostart/protected-apps allowlist. The stock Doze exemption above +// is well-known (though NOT officially documented anywhere by Google — +// verified against developer.android.com's Doze/App-Standby guide) to +// be insufficient on Xiaomi/Huawei/Honor/Oppo/Vivo/OnePlus, which layer +// their own aggressive killers behind a separate allowlist stock APIs +// cannot toggle. `needsOemAutostartSettings` only reports true when +// BOTH the manufacturer is one of those AND the OS confirms via the +// one official signal for this — `ActivityManager.isBackgroundRestricted` +// (API 28+) — that it's actually restricting this app right now; we +// deliberately don't nag every user on those OEMs unconditionally, only +// the ones the OS itself says are affected. `openOemAutostartSettings` +// opens the OEM screen (falling back to the app's standard settings +// page if no OEM-specific screen exists on this device/OS version). +// // All methods are safe no-ops on iOS and degrade gracefully on old Android // (the native side gates by API level). Failures are logged, never thrown — // nothing here may break pairing or the session flow. @@ -64,4 +78,76 @@ class AndroidBackground { debugPrint('[android-bg] battery-opt request failed: $e'); } } + + /// Manufacturers whose OS layers an aggressive process-killer on top of + /// stock Android Doze, gating background survival behind a separate + /// "autostart"/"protected apps" allowlist the stock + /// [requestIgnoreBatteryOptimizations] dialog does NOT cover. Lowercased + /// [Build.MANUFACTURER] substrings. + static const Set aggressiveOemManufacturers = { + 'xiaomi', + 'huawei', + 'honor', + 'oppo', + 'realme', + 'vivo', + 'oneplus', + }; + + /// The device manufacturer (lowercased, e.g. "xiaomi"), or null on iOS/error. + static Future manufacturerHint() async { + if (!Platform.isAndroid) return null; + try { + return await _ch.invokeMethod('manufacturerHint'); + } catch (e) { + debugPrint('[android-bg] manufacturer hint failed: $e'); + return null; + } + } + + /// Whether the OS is CURRENTLY restricting this app's background work — + /// `ActivityManager.isBackgroundRestricted()` (API 28+), the one official, + /// documented signal for this ("if true, any work that the app tries to do + /// will be aggressively restricted while it is in the background"). False + /// on iOS, on API <28, or on error (fails closed — never over-claims + /// restriction). + static Future isBackgroundRestricted() async { + if (!Platform.isAndroid) return false; + try { + return await _ch.invokeMethod('isBackgroundRestricted') == true; + } catch (e) { + debugPrint('[android-bg] background-restricted check failed: $e'); + return false; + } + } + + /// True when the extra OEM autostart step is actually worth offering: + /// this device's OEM is known to gate background survival behind a + /// separate allowlist AND the OS is presently confirmed to be restricting + /// this app (`isBackgroundRestricted`, API 28+) — i.e. we don't just guess + /// off the manufacturer string, we confirm against the one documented + /// signal for this situation before nagging the user. (API <28 predates + /// that signal entirely — `isBackgroundRestricted` reports false there, + /// so this option simply won't surface on those now-ancient devices; + /// acceptable given how old API <28 is at this point.) + static Future needsOemAutostartSettings() async { + final m = await manufacturerHint(); + if (m == null || !aggressiveOemManufacturers.any(m.contains)) return false; + return isBackgroundRestricted(); + } + + /// Open this OEM's autostart/protected-apps allowlist screen (a second, + /// stronger line of defense than the stock battery-optimization exemption + /// — see the native-side doc). Falls back to the app's standard "App info" + /// settings page if no OEM-specific screen exists on this device, so the + /// call always lands the user somewhere useful. No-op on iOS. + static Future openOemAutostartSettings() async { + if (!Platform.isAndroid) return; + try { + final outcome = await _ch.invokeMethod('openOemAutostartSettings'); + debugPrint('[android-bg] OEM autostart settings: $outcome'); + } catch (e) { + debugPrint('[android-bg] OEM autostart settings failed: $e'); + } + } } diff --git a/lib/ble/ble_engine.dart b/lib/ble/ble_engine.dart index 71b140fe..890f8971 100644 --- a/lib/ble/ble_engine.dart +++ b/lib/ble/ble_engine.dart @@ -36,6 +36,7 @@ // bursts), so the compute trigger survives the move to continuous listening. import 'dart:async'; +import 'dart:convert'; import 'dart:io'; import 'package:flutter/foundation.dart'; import 'package:flutter_blue_plus/flutter_blue_plus.dart'; @@ -601,8 +602,19 @@ class BleEngine { // caller pauses the auto-reconnect loop instead of pinning the radio forever. // A single successful bond clears it (see the createBond block below). final BondRefusalGiveUp _bondGiveUp = BondRefusalGiveUp(); + // Real per-chunk failure tracking (see ChunkFailureLedger doc) — persists + // across reconnects like marginal-radio/post-bond-loop/bond-give-up, since + // the whole point is catching the SAME token failing across sessions. + final ChunkFailureLedger _chunkFailures = ChunkFailureLedger(); EmptySyncTracker _emptySync = EmptySyncTracker(); StuckStrapDetector _stuckStrap = StuckStrapDetector(); + // Detector 1b: sustained frame corruption (CRC8/CRC32 failures) — an + // independent failure axis from marginal-radio, which only ever sees + // timeouts. Per-connection like empty-sync/stuck-strap: a new link starts + // with fresh radio conditions. + FrameCorruptionDetector _frameCorruption = FrameCorruptionDetector(); + int _crcFailuresTotal = 0; // across the engine's lifetime (diagnostics) + int _crcFailuresThisSession = 0; // reset on each connect ClockRef? _clockRef; // strap-RTC ↔ wall correlation (set from GET_CLOCK) /// Latest strap-RTC ↔ wall correlation, or null until GET_CLOCK is answered. @@ -612,8 +624,14 @@ class BleEngine { /// back, and the GET_CLOCK handler re-issues on drift — so cap the retries or /// a firmware that never latches either payload form would loop forever. int _clockCorrectTries = 0; + // Proactive RTC recheck timestamp for long-lived connections — see + // kRtcReverifyIntervalSeconds. Every other clock recheck is symptom-driven. + DateTime? _lastClockVerifyAt; int? _sessionOldestUnix; // strap's banked-data window (GET_DATA_RANGE) int? _sessionNewestUnix; + // Lifetime count of GET_DATA_RANGE reads rejected by isCorruptFutureRtc — + // see the range_oldest/range_newest handler below. + int _corruptDataRangeCount = 0; DateTime? _bondTime; // when the handshake completed (bond confirmed) DateTime? _armTime; // when live (R10/R11) streams were last armed int _autoContinueCount = 0; // consecutive auto-continues this connection @@ -624,11 +642,23 @@ class BleEngine { // EVERY historical-record path — see RecordGate in ble_state.dart. Re-seeded // on each connect from the durable cursor. RecordGate _recordGate = RecordGate(); + // Explicit, observable "band reboot" signal (see CounterRegressionDetector + // doc). Re-seeded from the durable counter_hw cursor on each connect, same + // pattern as _recordGate's frontierTs seed below. + CounterRegressionDetector _counterRegression = CounterRegressionDetector(); // Snapshot of `_recordGate.dropped` at the last HISTORY_START — lets the // HISTORY_END validator (below) tell "the band sent fewer packets than it // said" apart from "we correctly, silently rejected some as implausible // (stale-clock block) and never tallied them." See _handleSyncMarker. int _burstDroppedAtStart = 0; + // Band-truth reconciliation: `expectedPacketCount` mismatches are advisory + // (see the comment at the validation site — treating a single mismatch as + // fatal was actively harmful and was reverted), but a mismatch that keeps + // recurring burst after burst is a real signal worth surfacing over time + // rather than only as a single overwritten sync_ledger row. Pure + // observability — does NOT gate or retry anything. + int _burstMismatchTotal = 0; // across the engine's lifetime + int _burstMismatchStreak = 0; // consecutive mismatched bursts, reset by connect + by a clean burst // Per-revision packet accounting for the historical drain (gap detection + // honest per-version counts surfaced to the debug screens). final Map _historicalVersionCounts = {}; @@ -753,6 +783,21 @@ class BleEngine { // packet count before comparing against the band's expectedPacketCount. 'gate_dropped_total': _recordGate.dropped, 'gate_dropped_this_burst': _recordGate.dropped - _burstDroppedAtStart, + // CRC8/CRC32 frame failures — previously silent (see `_subscribe`). A + // rising count with a healthy `gate_dropped_*` is the signature of a + // degrading radio corrupting frames rather than a stale/implausible band. + 'crc_failures_total': _crcFailuresTotal, + 'crc_failures_this_session': _crcFailuresThisSession, + 'frame_corruption_tripped': _frameCorruption.tripped, + // Band-truth reconciliation: expectedPacketCount vs. what we actually + // committed — advisory only (see the comment at the validation site), but + // a *streak* of mismatches is a real signal worth watching over time. + 'burst_mismatch_total': _burstMismatchTotal, + 'burst_mismatch_streak': _burstMismatchStreak, + // Band-reboot signal — see CounterRegressionDetector. Observability only; + // recovery already happens automatically at the DB layer. + 'counter_regressions_total': _counterRegression.regressions, + 'corrupt_data_ranges_total': _corruptDataRangeCount, 'history_requests': _historyRequests, 'history_completions': _historyCompletions, 'successful_bursts': _successfulBursts, @@ -964,7 +1009,9 @@ class BleEngine { } _setPhase(BleConnState.discovering); - final services = await device.discoverServices(); + final services = await device + .discoverServices() + .timeout(_serviceDiscoveryTimeout); BluetoothService? svc; for (final s in services) { if (s.uuid.str.toLowerCase().startsWith('61080001')) svc = s; @@ -1007,11 +1054,15 @@ class BleEngine { // connect). Records stamped after this carry real unix time. _clockCorrectTries = 0; // fresh retry budget for this connection await setClock(); + _lastClockVerifyAt = DateTime.now(); // Per-connection policy reset. Marginal-radio + post-bond-loop are NOT reset // here — they count consecutive bad cycles across reconnects and self-reset on // a healthy disconnect. Empty-sync + stuck are per-connection. _emptySync = EmptySyncTracker(); _stuckStrap = StuckStrapDetector(); + _frameCorruption = FrameCorruptionDetector(); + _crcFailuresThisSession = 0; + _burstMismatchStreak = 0; _autoContinueCount = 0; _lastBackfillAt = 0; _successfulBursts = 0; @@ -1031,6 +1082,12 @@ class BleEngine { // continuation detectors are correct on the first offload after a restart. _recordGate = RecordGate(frontierTs: (await cursorReader?.call('rec_ts_hw')) ?? 0); + // Re-seed the counter-regression watch from the durable counter_hw + // cursor so a reboot is caught even across the reconnect it usually + // causes, instead of only within a single unbroken connection. + _counterRegression = CounterRegressionDetector( + seedCounter: await cursorReader?.call('counter_hw'), + ); // Heartbeat: keep the link alive (~10s LINK_VALID). Owned by the session, so a // disconnect cancels it — no zombie timer firing into a dead characteristic. @@ -1101,6 +1158,19 @@ class BleEngine { if (shouldPauseMaintenanceTraffic(offloadActive: _offloadActive)) { return; } + // Proactive RTC recheck: every other clock verification is symptom-driven + // (see kRtcReverifyIntervalSeconds doc). A long-lived link (e.g. iOS's + // bluetooth-central background mode, which can stay open indefinitely) + // gets an independent periodic GET_CLOCK; the existing clock_epoch + // response handler does the actual drift comparison + bounded re-issue. + final lastVerify = _lastClockVerifyAt; + if (lastVerify == null || + DateTime.now().difference(lastVerify).inSeconds >= + kRtcReverifyIntervalSeconds) { + _lastClockVerifyAt = DateTime.now(); + _log('[SYNC] Periodic RTC re-verify (long-lived connection).'); + unawaited(getClock()); + } if (_liveEnabled) { // Re-arm ONLY what the current live mode wants: re-sending the high-rate // R10/R11 toggle while in HR-only mode (background downgrade) or under the @@ -1208,14 +1278,33 @@ class BleEngine { BluetoothCharacteristic c, String role, ) async { - await c.setNotifyValue(true); + await c.setNotifyValue(true).timeout(_notifySetupTimeout); session.subs.add( c.onValueReceived.listen((chunk) { // Ignore notifications from a session we've already torn down. if (_session != session || !session.connected) return; _lastRx = DateTime.now(); for (final frame in session.asm[role]!.feed(chunk)) { - if (frame.valid) _onFrame(role, frame); + if (frame.valid) { + _onFrame(role, frame); + } else { + // Previously silent: a degrading radio corrupting frames looked + // identical to a healthy one everywhere. Now counted (surfaced in + // offloadSnapshot) and fed to an independent corruption-rate + // detector below, alongside RecordGate.dropped for plausibility + // rejections. + _crcFailuresTotal++; + _crcFailuresThisSession++; + } + if (_frameCorruption.feed(frame.valid)) { + state.standardHrFallback = true; + onState(state); + _log( + '[RECONNECT] frame-corruption tripped ' + '($_crcFailuresThisSession CRC failures this session) — ' + 'standard-HR fallback enabled.', + ); + } } }), ); @@ -1290,6 +1379,15 @@ class BleEngine { // and silently re-floods the same chunk forever. The per-write timeout stops // a hung write-with-response from stalling the whole write chain / drain. static const Duration _writeTimeout = Duration(seconds: 8); + // Every other step in the connect chain is timed (connect() itself: 20s, + // ACK writes: 8s). discoverServices()/setNotifyValue() previously had none — + // a wedged BLE stack here would hang connect() forever and never trip the + // outer catch-all that tears the session down, silently jamming the whole + // reconnect ladder above it (OS reconnect / restore-central / BG tasks never + // get a chance to help because we never reach a failure state). Timing these + // out lets the existing `catch (e)` in `_doConnect` do its job. + static const Duration _serviceDiscoveryTimeout = Duration(seconds: 15); + static const Duration _notifySetupTimeout = Duration(seconds: 15); Future _write(Uint8List raw) { final session = _session; @@ -1546,6 +1644,17 @@ class BleEngine { if (pt != PacketType.historicalData) return; final recType = frame.inner.length > 1 ? frame.inner[1] : -1; final counter = _counterFromInner(frame.inner); + // Explicit, observable band-reboot signal — see CounterRegressionDetector. + // 0 is _counterFromInner's fallback for a too-short frame, not a real + // counter value, so it's excluded to avoid a false regression report. + if (counter > 0 && _counterRegression.feed(counter)) { + _log( + '[SYNC] Record counter regressed (band likely rebooted): ' + 'counter=$counter, regressions_total=${_counterRegression.regressions}. ' + 'Recovery is automatic (REPLACE-by-rec_ts + orphan cascade) — this is ' + 'observability only.', + ); + } // Decode the record FIRST so we can stamp its REAL time onto rec_ts. The // DerivationEngine buckets/windows days by rec_ts, so a multi-day flash // backfill (all received in one sync) splits into correct per-real-day @@ -1707,11 +1816,29 @@ class BleEngine { } } if (f.containsKey('range_oldest') && f.containsKey('range_newest')) { - _sessionOldestUnix = f['range_oldest'] as int; - _sessionNewestUnix = f['range_newest'] as int; - state.dataRangeOldest = _sessionOldestUnix; - state.dataRangeNewest = _sessionNewestUnix; - onState(state); + final oldest = f['range_oldest'] as int; + final newest = f['range_newest'] as int; + // GET_DATA_RANGE responses are documented to occasionally carry junk at + // unstable offsets. Nothing previously sanity-checked `range_newest` + // before it tightened RecordGate's session window for the whole + // connection — a corrupt "newest" implausibly far in the future would + // silently poison that window. Reject and fall back to the broad + // absolute floor/ceiling instead. + if (isCorruptFutureRtc(newest, _wallSecs().round())) { + _corruptDataRangeCount++; + _log( + '[SYNC] GET_DATA_RANGE newest=$newest is implausibly far in the ' + 'future — treating as a corrupt strap RTC read; NOT tightening ' + 'this session\'s plausibility window ' + '(corrupt_ranges_total=$_corruptDataRangeCount).', + ); + } else { + _sessionOldestUnix = oldest; + _sessionNewestUnix = newest; + state.dataRangeOldest = oldest; + state.dataRangeNewest = newest; + onState(state); + } } if (d.kind == 'cmd_response' && f['hello'] is HelloInfo) { final h = f['hello'] as HelloInfo; @@ -1870,9 +1997,12 @@ class BleEngine { // progress, "last data" frozen indefinitely. Log the mismatch (still // useful signal — see the sync-diagnostics screen) and commit anyway. if (!validated) { + _burstMismatchTotal++; + _burstMismatchStreak++; _log( '[SYNC] Burst packet-count mismatch (advisory, NOT blocking commit) ' - '(attempt ${d.consecutiveValidationFailures}): expected=$expected, ' + '(attempt ${d.consecutiveValidationFailures}, ' + 'streak=$_burstMismatchStreak): expected=$expected, ' 'actual=${d.currentBurstPacketCount}, ' 'dropped_this_burst=$droppedThisBurst, ' 'historical=${d.currentBurstHistoricalPacketCount}, ' @@ -1892,6 +2022,8 @@ class BleEngine { 'burst_breakdown': d.currentBurstBreakdown, }, ); + } else { + _burstMismatchStreak = 0; } _successfulBursts++; _mergeValidatedBurst(d); @@ -1923,9 +2055,48 @@ class BleEngine { // link — the committed data is safe, and the next session's re-delivery // is dedup-safe (decoded rows REPLACE by rec_ts). if (!await _writeAckVerified(ack)) { + // Real per-chunk ledger row, keyed by the token itself — previously + // every ledger write here collapsed onto one shared 'capture' row, + // so a token that kept failing ACROSS reconnects (the "Groundhog + // Day" re-flood signature) left no trace distinguishing it from a + // one-off bounce. This does not change behavior — the bounce below + // is unconditional either way, and the data is already safe (durably + // committed above, before the ACK was ever attempted) — it only adds + // visibility, plus an explicit quarantine escalation once the SAME + // token has failed enough times to be a real, diagnosable problem. + final failCount = _chunkFailures.recordFailure(tokenHex); + await LocalDb.upsertSyncLedgerEntry( + chunkId: 'batch:$tokenHex', + kind: 'historical_batch', + status: 'ack_failed', + lastError: 'ack_write_exhausted', + metaPatch: { + 'batch_id': m.batchId, + 'records': d.records, + 'ack_failures': failCount, + }, + ); + if (_chunkFailures.shouldQuarantine(tokenHex)) { + await LocalDb.quarantineSyncChunk( + kind: 'historical_batch', + payloadJson: jsonEncode({ + 'token': tokenHex, + 'batch_id': m.batchId, + 'ack_failures': failCount, + }), + reason: 'persistent_ack_failure', + ); + _log( + '[SYNC] Batch token=$tokenHex has failed ACK $failCount times ' + 'across reconnects — quarantined for diagnosis. Data is safe ' + '(already committed); this only means the band has not yet ' + 'been told to trim, so it keeps re-sending the same batch.', + ); + } _log('[SYNC] BATCH-ACK FAILED after ' - '${ackRetryPolicy.maxAttempts} attempts (token=$tokenHex) — ' - 'bouncing the link; data is committed and the band will re-send.'); + '${ackRetryPolicy.maxAttempts} attempts (token=$tokenHex, ' + 'failures_for_this_token=$failCount) — bouncing the link; data ' + 'is committed and the band will re-send.'); unawaited( _teardownSession(intentional: false).then((_) { _setPhase(BleConnState.idle); // caller's reconnect loop takes over @@ -1933,6 +2104,7 @@ class BleEngine { ); return; } + _chunkFailures.recordSuccess(tokenHex); d.noteBatchAcked(); // ACKed and KEEP listening await LocalDb.upsertSyncLedgerEntry( status: 'acknowledged', @@ -1946,6 +2118,18 @@ class BleEngine { 'strap_history_newest_ts': _strapHistoryNewestTs, }, ); + // Same event, but a REAL per-chunk row keyed by the token — closes out + // whatever ack_failed history this token accumulated above. + await LocalDb.upsertSyncLedgerEntry( + chunkId: 'batch:$tokenHex', + kind: 'historical_batch', + status: 'acked', + ackedAt: DateTime.now().millisecondsSinceEpoch, + metaPatch: { + 'batch_id': m.batchId, + 'records': d.records, + }, + ); _noteStored(); // a banked batch → schedule a (debounced) derive } else if (m.sub == SyncMeta.historyComplete) { final d = _drain; diff --git a/lib/ble/ble_state.dart b/lib/ble/ble_state.dart index 8e6e30f0..2cc8a352 100644 --- a/lib/ble/ble_state.dart +++ b/lib/ble/ble_state.dart @@ -226,6 +226,48 @@ class RecordGate { } } +/// Explicit, observable signal for "the band's hardware record counter went +/// backwards" — the signature of a band reboot mid-offload (its onboard +/// counter resets). Recovery already happens correctly and silently at the +/// DB layer (`decoded_onehz` REPLACE-by-rec_ts + orphan-cascade delete on the +/// evicted counter's RR beats) — this adds NO new recovery behavior, only an +/// observable event, so a regression (and any future regression in how it's +/// handled) doesn't go unnoticed the way CRC failures used to before they +/// were counted. Seed [seedCounter] from the durable `counter_hw` cursor so a +/// regression is caught even across the reconnect that a reboot itself +/// usually causes — the two events are correlated, not sequential. +class CounterRegressionDetector { + int? _lastCounter; + + /// Regressions observed since construction (never reset by [reset] — reset + /// only clears the last-seen counter for reseeding at a fresh connect). + int regressions = 0; + + CounterRegressionDetector({int? seedCounter}) : _lastCounter = seedCounter; + + /// Feed the next record's raw hardware counter (u32, may wrap on a + /// sufficiently long-running band). Returns true exactly when this counter + /// is a genuine regression against the previous one (not benign u32 + /// wraparound near the top of the range). + bool feed(int counter) { + final prev = _lastCounter; + _lastCounter = counter; + if (prev == null || counter >= prev) return false; + // Wraparound guard: prev near the top of u32, counter near 0 is normal + // roll-over on an extremely long-running band, not a reboot. + const wrapGuard = 0xFFFFFFFF - 1000000; + if (prev >= wrapGuard && counter < 1000000) return false; + regressions++; + return true; + } + + /// Re-seed for a fresh connection (does not clear the lifetime [regressions] + /// count — that's diagnostics across the engine's lifetime). + void reseed(int? seedCounter) { + _lastCounter = seedCounter; + } +} + /// Pure retry schedule for the HISTORY_END batch-ACK write. /// /// The safe-trim invariant commits raw+samples+cursor DURABLY BEFORE the ACK, @@ -255,6 +297,42 @@ class AckRetryPolicy { } } +/// Tracks ACK-write failures per historical-batch token ACROSS RECONNECTS — +/// a chunk whose ACK keeps failing for the SAME token (the "Groundhog Day" +/// re-flood signature: the band never trims, so it re-sends the identical +/// batch next session) is a persistent, diagnosable problem distinct from a +/// one-off bounce. Pure counter + threshold; the caller owns actually +/// writing to sync_ledger/sync_quarantine and bouncing the link — which +/// already happens regardless of this class, since the data is safe either +/// way (durably committed before the ACK was ever attempted). This only adds +/// visibility into a chunk that's stuck, where previously nothing recorded +/// that the SAME token had failed before. +class ChunkFailureLedger { + final int quarantineThreshold; + ChunkFailureLedger({this.quarantineThreshold = 3}); + + final Map _failures = {}; + + /// Record another ACK failure for [tokenHex]. Returns the new failure count. + int recordFailure(String tokenHex) { + final n = (_failures[tokenHex] ?? 0) + 1; + _failures[tokenHex] = n; + return n; + } + + /// Current failure count for [tokenHex] (0 if never failed / already cleared). + int failureCount(String tokenHex) => _failures[tokenHex] ?? 0; + + /// Whether [tokenHex] has just crossed the quarantine threshold. + bool shouldQuarantine(String tokenHex) => + (_failures[tokenHex] ?? 0) >= quarantineThreshold; + + /// Clear tracking for [tokenHex] once it finally ACKs successfully. + void recordSuccess(String tokenHex) { + _failures.remove(tokenHex); + } +} + /// Pure debounce/coalesce logic for the "new data stored → derive" trigger. /// /// With continuous listening there is no discrete "sync done" signal, so we can't diff --git a/lib/compute/derivation_engine.dart b/lib/compute/derivation_engine.dart index 21b7e043..f86c31fe 100644 --- a/lib/compute/derivation_engine.dart +++ b/lib/compute/derivation_engine.dart @@ -21,6 +21,7 @@ import 'dart:async'; import 'dart:convert'; +import 'dart:io' show Platform; import 'dart:isolate'; import 'dart:math' as math; @@ -296,6 +297,55 @@ class _BaselineHistoryCache { /// older days simply aren't in the substrate and are naturally excluded. const int _rescanWindowDays = 21; +/// Per-day-local page/row accumulator for the prepare stage (see +/// `_prepareTargetDay`/`_loadSubstrateRange`). Deliberately NOT shared +/// `_diag` state — under concurrent per-day processing, multiple days +/// resetting/incrementing the same shared counters would race and produce +/// garbage diagnostics. Each day gets its own instance; only the final +/// per-day total is merged into the shared running max, once. +class _PrepareStats { + int pages = 0; + int rows = 0; +} + +/// Run [worker] over [items] with at most [concurrency] running at once. Each +/// of up to [concurrency] "lanes" pulls the next unclaimed item as soon as +/// it's free — a mix of fast (empty/mostly-empty day) and slow (heavy +/// backlog day) items keeps every lane continuously busy, rather than +/// lock-stepping in fixed-size batches where one slow item stalls an entire +/// batch. Pure orchestration: no DB/isolate awareness of its own — every +/// caller in this file catches errors INSIDE [worker] itself (a day that +/// fails is marked skipped and processing continues), so a throwing [worker] +/// is not part of the normal contract here, but note that (per +/// `Future.wait`'s default behavior) an uncaught throw would propagate out +/// and NOT stop already-in-flight sibling lanes from completing their +/// current item first. +/// +/// This is the ONE place run()/runDays()/rescanRecent() get their real, +/// multi-core parallelism from — replacing what used to be a fully +/// sequential `for` loop that left every core but one idle during a +/// multi-day backlog sweep. +@visibleForTesting +Future runWithConcurrency( + List items, + int concurrency, + Future Function(T item) worker, +) async { + if (items.isEmpty) return; + final poolSize = math.min(concurrency, items.length).clamp(1, items.length); + var nextIndex = 0; + Future lane() async { + while (true) { + final myIndex = nextIndex; + if (myIndex >= items.length) return; + nextIndex++; // no `await` since the read above — atomic claim + await worker(items[myIndex]); + } + } + + await Future.wait(List.generate(poolSize, (_) => lane())); +} + class DerivationEngine { DerivationEngine({this.log}); final void Function(String)? log; @@ -312,19 +362,18 @@ class DerivationEngine { 'duration_ms': null, 'raw_pages': 0, 'raw_rows': 0, - 'day_raw_pages': 0, - 'day_raw_rows': 0, 'max_day_raw_pages': 0, 'max_day_raw_rows': 0, - 'range_from_rec_ts': null, - 'range_to_rec_ts': null, 'scope_days': 0, 'scope_reason': null, 'prepared_days': 0, 'todo_days': 0, 'done_days': 0, 'skipped_days': 0, - 'active_day': null, + // List, not a single day — several days can be in flight concurrently + // (see run()'s bounded worker pool). + 'active_days': [], + 'concurrency': 1, 'last_error': null, }; @@ -354,19 +403,16 @@ class DerivationEngine { ..['duration_ms'] = null ..['raw_pages'] = 0 ..['raw_rows'] = 0 - ..['day_raw_pages'] = 0 - ..['day_raw_rows'] = 0 ..['max_day_raw_pages'] = 0 ..['max_day_raw_rows'] = 0 - ..['range_from_rec_ts'] = null - ..['range_to_rec_ts'] = null ..['scope_days'] = 0 ..['scope_reason'] = null ..['prepared_days'] = 0 ..['todo_days'] = 0 ..['done_days'] = 0 ..['skipped_days'] = 0 - ..['active_day'] = null + ..['active_days'] = [] + ..['concurrency'] = _deriveConcurrency ..['last_error'] = null; try { final scope = await _deriveScope(heavy: heavy, force: force); @@ -405,16 +451,37 @@ class DerivationEngine { : heavy ? "heavy" : "light"}; ' - '${scope.reason}; v$kAlgoVersion)', + '${scope.reason}; v$kAlgoVersion; ' + 'concurrency=$_deriveConcurrency)', ); + // Newest-first: `scope.targetDays` sorts ascending (oldest first), which + // is exactly backwards from what the user actually wants when they open + // the app after a backlog — today/most-recent should be among the very + // FIRST days dispatched, not the last one a long sweep gets to. A no-op + // for the light path (0-1 days), so always safe to apply. + final orderedDays = todoDays.reversed.toList(); + var done = 0; + var completed = 0; + final activeDays = {}; _diag['stage'] = 'per_day'; - for (var i = 0; i < todoDays.length; i++) { - final dayId = todoDays[i]; - _diag['active_day'] = dayId; + _diag['active_days'] = const []; + + // One day's full prepare→compute→persist body (identical to the old + // sequential loop's per-iteration work) — extracted so it can run as a + // unit inside the worker pool below. Concurrency-safe: everything it + // touches is either (a) day_id-keyed DB rows (independent across days), + // (b) the read-only `history` snapshot (frozen before this loop starts, + // refreshed only after it ends — see `_BaselineHistoryCache`), or (c) + // shared counters mutated via single, non-`await`-split statements, + // which Dart's cooperative single-threaded scheduler makes atomic + // relative to the other concurrent workers even though the actual + // isolate CPU work they await genuinely runs in parallel across cores. + Future processDay(String dayId) async { + activeDays.add(dayId); + _diag['active_days'] = activeDays.toList(); try { - _diag['stage'] = 'prepare'; final prepared = await _prepareTargetDay(dayId); // Override day whose raw has been pruned (≥14 d): re-deriving would // produce an empty/absent result and clobber the user's manual sleep. @@ -423,12 +490,8 @@ class DerivationEngine { prepared.daySub.isEmpty && overrideDays.contains(dayId)) { _log('derive day $dayId skipped: override day, raw pruned — kept'); - onDayDone?.call(dayId, i + 1, todoDays.length); - continue; - } - if (prepared != null) { + } else if (prepared != null) { _diag['prepared_days'] = (_diag['prepared_days'] as int) + 1; - _diag['stage'] = 'per_day'; await _derivePreparedDay(prepared, profile, dataNowSec, history); done++; _diag['done_days'] = done; @@ -455,9 +518,14 @@ class DerivationEngine { _diag['skipped_days'] = (_diag['skipped_days'] as int) + 1; _diag['last_error'] = '$e'; } - onDayDone?.call(dayId, i + 1, todoDays.length); + activeDays.remove(dayId); + _diag['active_days'] = activeDays.toList(); + completed++; + onDayDone?.call(dayId, completed, orderedDays.length); } + await runWithConcurrency(orderedDays, _deriveConcurrency, processDay); + // 4. Cross-day rollup + notifications (best-effort). if (done > 0) { _diag['stage'] = 'baselines'; @@ -483,7 +551,7 @@ class DerivationEngine { _diag ..['running'] = false ..['stage'] = 'idle' - ..['active_day'] = null + ..['active_days'] = const [] ..['finished_at'] = finishedAt ..['duration_ms'] = finishedAt - startedAt; } @@ -509,19 +577,16 @@ class DerivationEngine { ..['duration_ms'] = null ..['raw_pages'] = 0 ..['raw_rows'] = 0 - ..['day_raw_pages'] = 0 - ..['day_raw_rows'] = 0 ..['max_day_raw_pages'] = 0 ..['max_day_raw_rows'] = 0 - ..['range_from_rec_ts'] = null - ..['range_to_rec_ts'] = null ..['scope_days'] = days.length ..['scope_reason'] = 'selected-days' ..['prepared_days'] = 0 ..['todo_days'] = 0 ..['done_days'] = 0 ..['skipped_days'] = 0 - ..['active_day'] = null + ..['active_days'] = [] + ..['concurrency'] = _deriveConcurrency ..['last_error'] = null; try { final scope = _scopeForDays(days.toList(), reason: 'selected-days'); @@ -541,10 +606,17 @@ class DerivationEngine { } _diag['todo_days'] = todoDays.length; final history = await _BaselineHistoryCache.load(); + // Same bounded worker-pool pattern as run() — see its doc for why this + // is safe (independent day_id-keyed writes + a frozen baseline shared + // read-only across the whole batch). + final orderedDays = todoDays.reversed.toList(); var done = 0; - for (var i = 0; i < todoDays.length; i++) { - final dayId = todoDays[i]; - _diag['active_day'] = dayId; + var completed = 0; + final activeDays = {}; + + Future processDay(String dayId) async { + activeDays.add(dayId); + _diag['active_days'] = activeDays.toList(); try { final prepared = await _prepareTargetDay(dayId); if (prepared != null) { @@ -561,8 +633,13 @@ class DerivationEngine { _diag['skipped_days'] = (_diag['skipped_days'] as int) + 1; _diag['last_error'] = '$e'; } - onDayDone?.call(dayId, i + 1, todoDays.length); + activeDays.remove(dayId); + _diag['active_days'] = activeDays.toList(); + completed++; + onDayDone?.call(dayId, completed, orderedDays.length); } + + await runWithConcurrency(orderedDays, _deriveConcurrency, processDay); if (done > 0) { await _refreshBaselines(history); await _runCrossDay(profile); @@ -588,33 +665,49 @@ class DerivationEngine { static const int _maxDayRawPages = 300; Future _prepareTargetDay(String dayId) async { - _diag - ..['day_raw_pages'] = 0 - ..['day_raw_rows'] = 0; - _diag['stage'] = 'sleep_candidate'; - final candidate = await _sleepCandidateForDay(dayId); + // Per-day page/row totals used to live in the shared `_diag` map (reset + // then accumulated across this day's 2-3 substrate loads). Under + // concurrent per-day processing (see `run()`), multiple days resetting/ + // incrementing the SAME shared fields would race and produce garbage + // diagnostics (never a correctness issue for the derived VALUES — this + // is telemetry-only). Each day now gets its own local accumulator, + // merged into the shared running max exactly once, below. + final stats = _PrepareStats(); + final candidate = await _sleepCandidateForDay(dayId, stats: stats); final dayStart = _localDayLabelToSec(dayId); final dayEnd = dayStart + 86400; - _diag['stage'] = 'day_sub'; final daySub = await _loadSubstrateRange( dayStart, dayEnd - 1, dayId: dayId, + stats: stats, ); Substrate sleepSub = Substrate.empty; if (candidate.present && candidate.sleepOffsetSec > candidate.sleepOnsetSec) { - _diag['stage'] = 'sleep_sub'; sleepSub = await _loadSubstrateRange( candidate.sleepOnsetSec, candidate.sleepOffsetSec - 1, dayId: dayId, + stats: stats, ); } + // Single safe merge into the shared max-tracking diagnostics — one + // statement, no `await` in between, so it's atomic relative to any other + // concurrently-running day's identical merge. + if (stats.pages > (_diag['max_day_raw_pages'] as int)) { + _diag['max_day_raw_pages'] = stats.pages; + } + if (stats.rows > (_diag['max_day_raw_rows'] as int)) { + _diag['max_day_raw_rows'] = stats.rows; + } return candidate.toPreparedDay(daySub: daySub, sleepSub: sleepSub); } - Future _sleepCandidateForDay(String dayId) async { + Future _sleepCandidateForDay( + String dayId, { + _PrepareStats? stats, + }) async { // A user sleep override is the source of truth — never serve the cached auto // candidate, and don't cache the override result (so a later edit / clear is // not shadowed by a stale artifact). The auto path keeps its finalized cache. @@ -652,6 +745,7 @@ class DerivationEngine { range.$1, range.$2, dayId: dayId, + stats: stats, ); final candidate = prepareSleepSessionCandidate( searchSub, @@ -672,6 +766,7 @@ class DerivationEngine { int fromRecTs, int toRecTs, { required String dayId, + _PrepareStats? stats, }) async { if (toRecTs < fromRecTs) return Substrate.empty; final port = ReceivePort(); @@ -710,9 +805,6 @@ class DerivationEngine { int? afterCursor; var rangePages = 0; var rangeRows = 0; - _diag - ..['range_from_rec_ts'] = fromRecTs - ..['range_to_rec_ts'] = toRecTs; while (true) { final decodedRows = await LocalDb.decodedOneHzBatchByRecTsRange( limit: _rawDecodeBatchSize, @@ -725,6 +817,10 @@ class DerivationEngine { _trackPrepareBatch(decodedRows.length); rangePages += 1; rangeRows += decodedRows.length; + if (stats != null) { + stats.pages += 1; + stats.rows += decodedRows.length; + } _enforcePrepareBudget( dayId: dayId, fromRecTs: fromRecTs, @@ -759,17 +855,14 @@ class DerivationEngine { } } + // Cumulative across the WHOLE run — safe under concurrent per-day + // processing since each field is a simple, non-`await`-split increment + // (order across days doesn't matter for a total). Per-day max tracking + // moved to `_PrepareStats` + the single merge at the end of + // `_prepareTargetDay`, since that DOES need per-day isolation. void _trackPrepareBatch(int rows) { _diag['raw_pages'] = (_diag['raw_pages'] as int) + 1; _diag['raw_rows'] = (_diag['raw_rows'] as int) + rows; - _diag['day_raw_pages'] = (_diag['day_raw_pages'] as int) + 1; - _diag['day_raw_rows'] = (_diag['day_raw_rows'] as int) + rows; - if ((_diag['day_raw_pages'] as int) > (_diag['max_day_raw_pages'] as int)) { - _diag['max_day_raw_pages'] = _diag['day_raw_pages']; - } - if ((_diag['day_raw_rows'] as int) > (_diag['max_day_raw_rows'] as int)) { - _diag['max_day_raw_rows'] = _diag['day_raw_rows']; - } } void _enforcePrepareBudget({ @@ -897,22 +990,32 @@ class DerivationEngine { ); final history = await _BaselineHistoryCache.load(); + // Same bounded worker-pool pattern as run()/runDays — up to + // _rescanWindowDays (21) days is exactly the kind of sweep that used + // to run fully sequentially for no reason (independent day_id-keyed + // writes + one frozen baseline snapshot shared read-only here). + final orderedDays = todoDays.reversed.toList(); var done = 0; - for (var i = 0; i < todoDays.length; i++) { - final dayId = todoDays[i]; + var completed = 0; + + Future processDay(String dayId) async { try { final prepared = await _prepareTargetDay(dayId); - if (prepared == null) continue; - await _derivePreparedDay(prepared, profile, dataNowSec, history); - done++; + if (prepared != null) { + await _derivePreparedDay(prepared, profile, dataNowSec, history); + done++; + } } catch (e) { _log('rescan day $dayId FAILED/skipped: $e'); // Do NOT mark-skipped here — a finalized day already has a good row; // overwriting it with a skip marker would DISCARD real structure. } - onDayDone?.call(dayId, i + 1, todoDays.length); + completed++; + onDayDone?.call(dayId, completed, orderedDays.length); } + await runWithConcurrency(orderedDays, _deriveConcurrency, processDay); + await _refreshBaselines(history); // Cross-day rollup + notifications reflect the refreshed scalars. await _runCrossDay(profile); @@ -958,6 +1061,31 @@ class DerivationEngine { /// skipped so the sweep always makes progress. static const Duration _perDayTimeout = Duration(seconds: 90); + /// Bounded worker-pool size for concurrent per-day derivation. Days within + /// a single run share ONE frozen baseline snapshot (`_BaselineHistoryCache` + /// is loaded once before the loop, refreshed once after — see `run()`) and + /// each writes to an independent, day_id-keyed `day_result` row — there is + /// no cross-day ordering dependency within a run. A multi-day backlog sweep + /// was previously fully sequential (one day's prepare-isolate + prepare + /// substrate loads + compute-isolate all finishing before the next day even + /// started), which wastes every core beyond the one doing the current day's + /// work. Running several days' isolate work genuinely concurrently gets + /// real wall-clock speedup from the device's other cores. Capped + /// conservatively — this is a phone doing background/foreground compute, + /// not a server batch job — rather than using every available core. + static const int _maxDeriveConcurrency = 3; + + int get _deriveConcurrency { + try { + return math.max( + 1, + math.min(_maxDeriveConcurrency, Platform.numberOfProcessors), + ); + } catch (_) { + return 1; // Platform unavailable on this target — sequential fallback + } + } + // ── imports (derive from a pre-built substrate, not from stored raw) ───────── /// Derive the named [dates] from a caller-supplied [sub] (e.g. a CSV import diff --git a/lib/data/db.dart b/lib/data/db.dart index d0dd43f6..b1547205 100644 --- a/lib/data/db.dart +++ b/lib/data/db.dart @@ -565,13 +565,30 @@ class LocalDb { /// continuation cursor in ONE transaction. This is the durable half of the /// safe-trim invariant — it MUST return before the engine writes the ACK frame. /// Advances counter_hw / rec_ts_hw to the batch max so a restart resumes cleanly. + /// + /// [onCheckpoint], if given, is called synchronously at each of the three + /// phases (decoded+archive queued, decoded+archive committed, cursor + /// advanced) — field diagnosability without giving up the single-txn + /// atomicity: the checkpoint calls are pure logging, wrapped so a + /// misbehaving callback can never abort a real commit. db.dart itself stays + /// logging-framework-free (no Flutter dependency); callers pass their own + /// logger (e.g. ble_engine.dart's `_log`, background_sync.dart's `debugPrint`). static Future commitSyncBatch( List raws, List samples, { String? trimToken, Map? extraCursors, List? archives, + void Function(String)? onCheckpoint, }) async { + void checkpoint(String msg) { + try { + onCheckpoint?.call(msg); + } catch (_) { + /* a logging callback must never affect the commit */ + } + } + final db = await instance; await db.transaction((txn) async { // Read the existing high-water THROUGH the txn — never via the global db @@ -608,7 +625,12 @@ class LocalDb { if (raw.counter > maxCounter) maxCounter = raw.counter; if (recTs > maxRecTs) maxRecTs = recTs; } + checkpoint( + 'decoded_archive_queued raws=${raws.length} ' + 'archives=${archives?.length ?? 0}', + ); await batch.commit(noResult: true); + checkpoint('decoded_archive_committed'); await setCursor('counter_hw', '$maxCounter', txn: txn); await setCursor('rec_ts_hw', '$maxRecTs', txn: txn); if (trimToken != null) await setCursor('strap_trim', trimToken, txn: txn); @@ -617,6 +639,10 @@ class LocalDb { await setCursor(e.key, e.value, txn: txn); } } + checkpoint( + 'cursor_advanced counter_hw=$maxCounter rec_ts_hw=$maxRecTs ' + 'trim=${trimToken != null}', + ); }); await _writeCaptureFreshness(raws); } @@ -1930,6 +1956,17 @@ class LocalDb { }); } + /// Previously write-only: `quarantineSyncChunk` had no reader anywhere, + /// so a persistently-stuck batch was recorded but never actually + /// retrievable for diagnosis. Append-only audit trail (like `raw_archive`) + /// — never pruned here, resolution is implicit once the same token finally + /// ACKs (see the `batch:$tokenHex` sync_ledger row transitioning to + /// `acked`), not a deletion of the quarantine record. + static Future>> quarantinedSyncChunks() async { + final db = await instance; + return db.query('sync_quarantine', orderBy: 'created_at DESC'); + } + static Future> samplesInRange(int fromTs, int toTs) async { final db = await instance; final decodedRows = await db.query( diff --git a/lib/notify/notification_center.dart b/lib/notify/notification_center.dart index 13a1aeab..4e820165 100644 --- a/lib/notify/notification_center.dart +++ b/lib/notify/notification_center.dart @@ -22,7 +22,21 @@ class NotificationCenter { static final NotificationCenter instance = NotificationCenter._(); /// Persist to the feed and (if allowed) present to the OS. Never throws. - Future emit(NotificationEvent e) async { + /// + /// [allowPermissionPrompt]: Apple's notification docs document that + /// authorization must be requested IN CONTEXT, from an active foreground + /// scene — never from a background execution context (a headless + /// BGTaskScheduler run or Dart background isolate has none to present + /// from). Callers that know they're running headless (see + /// background_sync.dart's checkSyncStaleness) MUST pass `false`, so a + /// not-yet-decided permission is checked, not requested, and never gets + /// permanently mis-cached as "denied" by a background attempt. The in-app + /// feed write above is unaffected either way — it's always written, + /// independent of OS permission. + Future emit( + NotificationEvent e, { + bool allowPermissionPrompt = true, + }) async { final nowMs = DateTime.now().millisecondsSinceEpoch; // Feed first — INSERT OR IGNORE keyed on dedupeKey, so re-runs don't dup. try { @@ -34,7 +48,10 @@ class NotificationCenter { final now = DateTime.now(); final minuteOfDay = now.hour * 60 + now.minute; if (prefs.shouldFireOs(e, minuteOfDay)) { - await NotificationService.instance.presentEvent(e); + await NotificationService.instance.presentEvent( + e, + allowPermissionPrompt: allowPermissionPrompt, + ); } } catch (_) {/* OS present best-effort */} } diff --git a/lib/notify/notification_service.dart b/lib/notify/notification_service.dart index 6140b950..2e265b04 100644 --- a/lib/notify/notification_service.dart +++ b/lib/notify/notification_service.dart @@ -146,9 +146,25 @@ class NotificationService { } /// Request notification permission once (iOS always; Android 13+). Cached. - Future ensurePermission() async { + /// + /// [allowPrompt] gates whether this may show the OS's interactive + /// authorization dialog. Apple's notification docs ("Asking permission to + /// use notifications") document that authorization should be requested in + /// CONTEXT — the interactive prompt assumes an active foreground scene to + /// present from — never automatically, and never from a background + /// execution context (a headless BGTaskScheduler/BGAppRefreshTask run or a + /// Dart background isolate has no such scene). Callers that know they're + /// running headless (see background_sync.dart's checkSyncStaleness) must + /// pass `allowPrompt: false`; every foreground/contextual caller keeps the + /// default `true`. With `false` and no prior decision cached, this checks + /// (never requests) via `checkPermissions()` and fails closed to `false` + /// rather than attempting to prompt — matching the "in-app feed is ALWAYS + /// written, OS presentation is best-effort" contract in NotificationCenter. + Future ensurePermission({bool allowPrompt = true}) async { await init(); if (_granted != null) return _granted!; + if (!allowPrompt) return hasPermission(); + bool granted = true; final ios = _plugin.resolvePlatformSpecificImplementation< IOSFlutterLocalNotificationsPlugin>(); @@ -166,6 +182,28 @@ class NotificationService { return granted; } + /// Non-mutating: whether notifications are currently enabled, WITHOUT ever + /// showing the OS authorization prompt. Safe to call from any context, + /// including headless/background. Does not populate [_granted] — a + /// not-yet-decided status here shouldn't get permanently cached as + /// "denied" just because a background check happened to run first. + Future hasPermission() async { + try { + await init(); + final ios = _plugin.resolvePlatformSpecificImplementation< + IOSFlutterLocalNotificationsPlugin>(); + if (ios != null) return (await ios.checkPermissions())?.isEnabled ?? false; + final android = _plugin.resolvePlatformSpecificImplementation< + AndroidFlutterLocalNotificationsPlugin>(); + if (android != null) { + return await android.areNotificationsEnabled() ?? false; + } + return true; // other platforms (macOS/Linux) — no gating here + } catch (_) { + return false; + } + } + NotificationDetails _details(NotifCategory c) { final ch = _channelFor(c); return NotificationDetails( @@ -183,9 +221,15 @@ class NotificationService { /// Present a NotificationEvent on its category channel. Same osId replaces, so /// re-firing the same logical event never stacks duplicates. Never throws. - Future presentEvent(NotificationEvent e) async { + /// + /// [allowPermissionPrompt] — see [ensurePermission]'s doc. Pass `false` from + /// any caller that knows it's running headless/in the background. + Future presentEvent( + NotificationEvent e, { + bool allowPermissionPrompt = true, + }) async { try { - if (!await ensurePermission()) return; + if (!await ensurePermission(allowPrompt: allowPermissionPrompt)) return; await _plugin.show( e.osId, e.title, diff --git a/lib/state/app_state.dart b/lib/state/app_state.dart index 400b486f..400f6f8e 100644 --- a/lib/state/app_state.dart +++ b/lib/state/app_state.dart @@ -57,12 +57,14 @@ import '../notify/notification_relay.dart'; import '../notify/notification_service.dart'; import '../notify/tap_router.dart'; import '../notify/water_buzzer.dart'; +import '../sync/background_sync.dart' show checkSyncStaleness; import '../sync/edge_tracking.dart'; import '../sync/band_ownership.dart'; import '../sync/high_freq_wake_window.dart'; import '../sync/ios_bg_task.dart'; import '../sync/paired_device.dart'; -import '../sync/sync_policy.dart' show isLinkStale; +import '../sync/sync_policy.dart' + show isLinkStale, StalenessTier, stalenessTierFor; import '../sync/update_service.dart'; import '../telemetry/telemetry_service.dart'; import '../telemetry/health_uploader.dart'; @@ -600,7 +602,9 @@ class AppState extends ChangeNotifier { // from the durable high-water on (re)connect. onCommitBatch: (raws, samples, trimTokenHex, {archives}) => LocalDb.commitSyncBatch(raws, samples, - trimToken: trimTokenHex, archives: archives), + trimToken: trimTokenHex, + archives: archives, + onCheckpoint: (msg) => _log('[COMMIT] $msg')), // Pre-setup fallback only: the drain path archives inside commitSyncBatch. onArchiveRecord: LocalDb.archiveRawRecord, cursorReader: LocalDb.getCursorInt, @@ -793,6 +797,16 @@ class AppState extends ChangeNotifier { await _maybeNotifyStepGoal(); await _maybeNotifyInactivity(); await _maybeGenerateBriefing(); + unawaited(_checkSchemaHealth()); // throttled internally to 24h + // Staleness-escalation meta-layer: the SAME check the headless path + // runs (shared cooldown via SharedPreferences, so foreground and + // background never double-fire) — a foreground open is exactly when a + // background wake-source failure streak should finally surface. + // allowPermissionPrompt:true is correct HERE (unlike the headless + // default) — runCadenceChecks only ever runs from an active foreground + // scene (app.dart's didChangeAppLifecycleState), so this is a genuinely + // contextual moment to ask, per Apple's/Android's notification docs. + unawaited(checkSyncStaleness(allowPermissionPrompt: true)); } catch (e) { _log('[notify] cadence checks skipped: $e'); } @@ -1143,6 +1157,8 @@ class AppState extends ChangeNotifier { unawaited(gestureSettings.bootstrap()); // Notification relay (Android only; inert + invisible elsewhere). Best-effort. unawaited(notificationRelay.bootstrap()); + // DB integrity check — see _checkSchemaHealth doc. Best-effort, non-blocking. + unawaited(_checkSchemaHealth()); initialized = true; notifyListeners(); // Companion (anonymous telemetry + health-data contribution) — best-effort, @@ -1205,6 +1221,37 @@ class AppState extends ChangeNotifier { if (logLines.length > 200) logLines.removeLast(); } + /// `LocalDb.schemaHealth()` (real `PRAGMA integrity_check` + schema + /// presence check) was previously fully implemented but never called + /// anywhere in the app — corruption or schema drift could accumulate + /// silently forever with nothing to notice it. Wired here: once at + /// startup, and at most once per [_schemaHealthCheckInterval] thereafter + /// via the existing foreground cadence (runCadenceChecks) so it doesn't + /// need its own timer infrastructure. Best-effort, never blocks boot. + static const Duration _schemaHealthCheckInterval = Duration(hours: 24); + Map? schemaHealth; + DateTime? _lastSchemaHealthCheckAt; + + Future _checkSchemaHealth({bool force = false}) async { + final last = _lastSchemaHealthCheckAt; + if (!force && + last != null && + DateTime.now().difference(last) < _schemaHealthCheckInterval) { + return; + } + _lastSchemaHealthCheckAt = DateTime.now(); + try { + final health = await LocalDb.schemaHealth(); + schemaHealth = health; + if (health['ok'] != true) { + _log('[db] schemaHealth FAILED: $health'); + } + notifyListeners(); + } catch (e) { + _log('[db] schemaHealth check skipped: $e'); + } + } + void _bumpInsightsRevision() { insightsRevision.value = insightsRevision.value + 1; } @@ -1760,6 +1807,18 @@ class AppState extends ChangeNotifier { Future requestIgnoreBatteryOptimizations() => AndroidBackground.requestIgnoreBatteryOptimizations(); + /// True when this device's OEM (Xiaomi/Huawei/Honor/Oppo/Vivo/OnePlus) is + /// known to gate background survival behind an extra autostart/protected- + /// apps allowlist the stock battery-optimization exemption doesn't cover. + /// Always false on iOS. + Future needsOemAutostartSettings() => + AndroidBackground.needsOemAutostartSettings(); + + /// Open this OEM's autostart allowlist screen (falls back to the app's + /// standard settings page if none exists on this device). + Future openOemAutostartSettings() => + AndroidBackground.openOemAutostartSettings(); + Future unpair() async { _keepAlive = false; BandOwnership.markForegroundIntent(false); @@ -2303,6 +2362,18 @@ class AppState extends ChangeNotifier { ? null : DateTime.fromMillisecondsSinceEpoch(_lastRecTs! * 1000); + /// The quiet in-app half of the staleness-escalation meta-layer (see + /// sync_policy.dart's stalenessTierFor doc + checkSyncStaleness, which + /// drives the louder OS-notification half). A never-synced band reads as + /// [StalenessTier.fresh] — that's a distinct, already-visible onboarding + /// state, not silent staleness. Always computed fresh against wall-clock + /// now, so any screen can read it without needing its own refresh timer. + StalenessTier get syncStalenessTier { + final last = lastRecordAt; + if (last == null) return StalenessTier.fresh; + return stalenessTierFor(DateTime.now().difference(last).inSeconds); + } + void _setBusy(bool b) { busy = b; notifyListeners(); diff --git a/lib/sync/background_sync.dart b/lib/sync/background_sync.dart index 8d14fcb3..81445aa1 100644 --- a/lib/sync/background_sync.dart +++ b/lib/sync/background_sync.dart @@ -18,9 +18,12 @@ import '../ble/ble_engine.dart'; import '../compute/derivation_engine.dart'; import '../compute/profile.dart'; import '../data/db.dart'; +import '../notify/notification_center.dart'; +import '../notify/notification_event.dart'; import 'band_ownership.dart'; import 'high_freq_wake_window.dart'; import 'paired_device.dart'; +import 'sync_policy.dart'; /// Load the local profile (no Provider in the headless isolate). Future _loadProfile() async { @@ -68,7 +71,9 @@ Future runHeadlessSync({BandLease? lease}) async { onRecordsBatch: LocalDb.insertRecordsBatch, onCommitBatch: (raws, samples, trimTokenHex, {archives}) => LocalDb.commitSyncBatch(raws, samples, - trimToken: trimTokenHex, archives: archives), + trimToken: trimTokenHex, + archives: archives, + onCheckpoint: (msg) => debugPrint('[bgsync][COMMIT] $msg')), onArchiveRecord: LocalDb.archiveRawRecord, cursorReader: LocalDb.getCursorInt, // Mark this as the background drainer: if the foreground app engine already @@ -85,6 +90,7 @@ Future runHeadlessSync({BandLease? lease}) async { debugPrint( '[bgsync] strap not reachable this cycle — will catch up next time.', ); + await checkSyncStaleness(); return true; } try { @@ -122,6 +128,7 @@ Future runHeadlessSync({BandLease? lease}) async { debugPrint('[bgsync] derive skipped: $e'); } debugPrint('[bgsync] done (local drain + light derive).'); + await checkSyncStaleness(); return true; } catch (e) { debugPrint('[bgsync] error (ignored): $e'); @@ -134,3 +141,59 @@ Future runHeadlessSync({BandLease? lease}) async { BandOwnership.release(ownedLease); } } + +// ── staleness escalation (meta-layer over the whole reconnect/sync ladder) ── +// See sync_policy.dart's stalenessTierFor doc. Evaluated at the end of every +// headless cycle (success OR a failed connect attempt — both are meaningful +// signals here), independent of THIS cycle's outcome: it reads the durable +// `rec_ts_hw` cursor, which reflects the full sync history, not just this run. +const String _kLastStalenessNotifiedMs = 'last_staleness_notified_ms'; + +/// [allowPermissionPrompt] defaults to `false` because this function's +/// PRIMARY callers (below, inside [runHeadlessSync]) run headless — see +/// NotificationCenter.emit's doc on why a background context must never +/// trigger the OS's interactive authorization prompt. app_state.dart's +/// foreground call (via runCadenceChecks, a genuinely contextual moment) +/// passes `true` explicitly. +Future checkSyncStaleness({bool allowPermissionPrompt = false}) async { + try { + final recTsHw = await LocalDb.getCursorInt('rec_ts_hw'); + // Never synced at all (e.g. freshly paired, first drain still pending) — + // nothing to escalate; that's a distinct, already-visible onboarding + // state, not silent staleness. + if (recTsHw == null || recTsHw <= 0) return; + final nowSec = DateTime.now().millisecondsSinceEpoch ~/ 1000; + final tier = stalenessTierFor(nowSec - recTsHw); + if (tier != StalenessTier.notify) return; + + final prefs = await SharedPreferences.getInstance(); + final lastMs = prefs.getInt(_kLastStalenessNotifiedMs); + final lastAt = + lastMs == null ? null : DateTime.fromMillisecondsSinceEpoch(lastMs); + final now = DateTime.now(); + if (!shouldRenotifyStaleness(lastAt, now)) return; + + await prefs.setInt(_kLastStalenessNotifiedMs, now.millisecondsSinceEpoch); + final hoursStale = (nowSec - recTsHw) ~/ 3600; + await NotificationCenter.instance.emit( + NotificationEvent( + // Date-bucketed so a legitimate re-fire after the cooldown isn't + // blocked by putNotification's INSERT-OR-IGNORE dedupe. + dedupeKey: '${now.toIso8601String().substring(0, 10)}:sync_stale', + category: NotifCategory.device, + priority: NotifPriority.normal, // respects quiet hours — not urgent + title: "Your band hasn't synced in a while", + body: 'No new data for about $hoursStale hours. Open OpenStrap to ' + 'reconnect — background sync may have stalled.', + date: now.toIso8601String().substring(0, 10), + route: '/today', + ), + allowPermissionPrompt: allowPermissionPrompt, + ); + debugPrint( + '[bgsync] staleness notification fired (hours_stale=$hoursStale).', + ); + } catch (e) { + debugPrint('[bgsync] staleness check skipped: $e'); + } +} diff --git a/lib/sync/headless_gate.dart b/lib/sync/headless_gate.dart index c727d085..69911bc3 100644 --- a/lib/sync/headless_gate.dart +++ b/lib/sync/headless_gate.dart @@ -20,25 +20,62 @@ class HeadlessSyncGate { HeadlessSyncGate._(); static Future? _running; + static String? _runningOwner; /// True while any headless entry point holds the gate. static bool get busy => _running != null; + // Skip-streak telemetry. Previously a collision between wake sources was + // only ever a single debugPrint line with no counter — a wake source that + // keeps losing the race EVERY cycle (a sign two sources are colliding + // rather than actually diversifying background coverage, e.g. the + // BLE-restore wake and a BGAppRefreshTask firing back-to-back every time) + // looked identical to one that skipped once by chance. Per-owner + // consecutive-skip + lifetime-total counters make that pattern observable. + static final Map _consecutiveSkipsByOwner = {}; + static int _totalSkips = 0; + + /// Consecutive skips for [owner] since it last actually ran (0 if it ran + /// most recently, or has never skipped). + static int consecutiveSkipsFor(String owner) => + _consecutiveSkipsByOwner[owner] ?? 0; + + /// Lifetime skip count across every owner (process-wide, resets on relaunch). + static int get totalSkips => _totalSkips; + /// Run [body] exclusively. If another entry point already holds the gate the /// call is SKIPPED (returns null) — same "skip, don't queue" semantics the /// old per-flag guards had, but shared across all entry points. static Future tryRun(String owner, Future Function() body) async { if (_running != null) { - debugPrint('[headless-gate] busy — "$owner" skipped this cycle'); + final n = (_consecutiveSkipsByOwner[owner] ?? 0) + 1; + _consecutiveSkipsByOwner[owner] = n; + _totalSkips++; + debugPrint( + '[headless-gate] busy (held by "$_runningOwner") — "$owner" skipped ' + 'this cycle (consecutive_skips=$n, total_skips=$_totalSkips)', + ); return null; } + _consecutiveSkipsByOwner[owner] = 0; // this run breaks its own streak final done = Completer(); _running = done.future; + _runningOwner = owner; try { return await body(); } finally { _running = null; + _runningOwner = null; done.complete(); } } + + /// Test-only reset — static state otherwise leaks across test cases. + @visibleForTesting + static void resetForTest() { + _running = null; + _runningOwner = null; + _consecutiveSkipsByOwner.clear(); + _totalSkips = 0; + } } diff --git a/lib/sync/sync_policy.dart b/lib/sync/sync_policy.dart index 1e877b5e..d219b84c 100644 --- a/lib/sync/sync_policy.dart +++ b/lib/sync/sync_policy.dart @@ -1,12 +1,14 @@ // sync_policy.dart — pure, I/O-free reconnect/offload policy. -// Value-typed state machines covering the six detectors + BackfillPolicy + -// clock/plausibility gates. NOTHING here touches BLE, the DB, or Flutter — -// every type is exhaustively unit-testable and the engine just feeds it -// observations and reads back decisions. +// Value-typed state machines covering the reconnect/offload detectors + +// BackfillPolicy + clock/plausibility gates. NOTHING here touches BLE, the +// DB, or Flutter — every type is exhaustively unit-testable and the engine +// just feeds it observations and reads back decisions. // -// WHOOP 4.0 only (the only family OpenStrap supports). The WHOOP-5-specific -// Whoop5EmptyOffloadTracker is included for completeness but is not wired by -// the engine. +// WHOOP 4.0 only (the only family OpenStrap supports). A prior speculative +// Whoop5EmptyOffloadTracker was removed — unwired, untested dead code with no +// real WHOOP5 offload path to integrate with. Write its real equivalent +// alongside actual WHOOP5 support when that's built; it won't be identical +// to the speculative shape anyway once real integration requirements exist. import 'dart:math' as math; @@ -16,6 +18,15 @@ const int kKeepAliveIntervalSeconds = 30; // re-arm realtime, poll battery, watchdog const int kBackfillIdleTimeoutSeconds = 60; // strap went silent mid-offload const int kLivenessFuseSeconds = 120; // no data for >fuse ⇒ bounce the link +// Every existing RTC recheck is symptom-triggered (drift detected on the ONE +// GET_CLOCK read at connect, or a defensive SET_CLOCK on StuckStrapDetector +// tripping). A connection that stays open for many hours (iOS's +// bluetooth-central background mode keeps links open indefinitely) had NO +// independent proactive recheck. This re-arms a plain GET_CLOCK on a healthy +// long-lived link; the existing clock_epoch response handler already does +// the drift comparison + bounded SET_CLOCK re-issue, so this only supplies +// the missing periodic trigger, not new drift-correction logic. +const int kRtcReverifyIntervalSeconds = 6 * 3600; // 6h const int kHistoricalSendFloorSeconds = 5; // official app floors 0x16 to 5s const int kHistoricalAbortRetryDelaySeconds = 3; // official app retries 3s after abort @@ -67,6 +78,19 @@ bool isPlausibleUnix( return true; } +/// A stricter, single-purpose sanity check for a band-reported "newest +/// record" timestamp (GET_DATA_RANGE's `range_newest`) that sits implausibly +/// far in the future — the signature of a corrupt/never-set strap RTC. +/// Deliberately narrower than [isPlausibleUnix] (which also enforces the +/// historical floor and the strap's own ±7-day session band): this is an +/// EARLY gate on the raw band-reported value itself, before it's trusted +/// enough to tighten the per-record plausibility window for the rest of the +/// session — GET_DATA_RANGE responses are documented to occasionally carry +/// junk at unstable offsets, and today nothing sanity-checks them before they +/// feed [RecordGate]'s session window. +bool isCorruptFutureRtc(int reportedUnix, int wallNow) => + reportedUnix > wallNow + kFutureMargin; + // ── clock correlation ──────────────────────────────────────────────────────── /// Correlates the strap's RTC epoch with wall-clock at the instant of GET_CLOCK. class ClockRef { @@ -263,6 +287,57 @@ class MarginalRadioDetector { } } +// ── detector 1b: frame corruption ──────────────────────────────────────────── +/// [MarginalRadioDetector] only ever sees *timeouts* — a link that stops +/// responding. A degrading radio can instead keep responding promptly while +/// corrupting an increasing fraction of frames (CRC8 length-byte or CRC32 +/// payload mismatch in `framing.dart`) — which looks identical to a healthy +/// link on every timeout-based diagnostic. This tracks a rolling window of +/// frame validity and trips once the corruption rate within the window +/// crosses [rateThreshold], driving the SAME standard-HR fallback action as +/// marginal-radio — same remedy (drop the raw live stream, keep 1 Hz decode), +/// independent failure signature. [minSamples] avoids tripping on a tiny, +/// noisy sample right after connect. +class FrameCorruptionDetector { + final int windowSize; + final double rateThreshold; + final int minSamples; + + FrameCorruptionDetector({ + this.windowSize = 50, + this.rateThreshold = 0.2, + this.minSamples = 20, + }); + + final List _window = []; // true = valid frame + int _invalidInWindow = 0; + bool tripped = false; + + /// Feed one frame's validity. Returns true exactly once, on the call that + /// crosses the threshold. + bool feed(bool valid) { + _window.add(valid); + if (!valid) _invalidInWindow++; + if (_window.length > windowSize) { + final removed = _window.removeAt(0); + if (!removed) _invalidInWindow--; + } + if (tripped || _window.length < minSamples) return false; + final rate = _invalidInWindow / _window.length; + if (rate >= rateThreshold) { + tripped = true; + return true; + } + return false; + } + + void reset() { + _window.clear(); + _invalidInWindow = 0; + tripped = false; + } +} + // ── detector 2: post-bond timeout loop (#617) ──────────────────────────────── /// Bond succeeds then dies ~1s later, re-scans, repeats. Consecutive /// bond→quick-timeout cycles. Trips once; action = surface the re-pair guide. @@ -372,34 +447,6 @@ class EmptySyncTracker { void reset() => _consecutive = 0; } -// ── detector 4: WHOOP-5 empty offload (#580) — NOT wired (no WHOOP5) ────────── -class Whoop5EmptyOffloadTracker { - final int quietThreshold; - Whoop5EmptyOffloadTracker({this.quietThreshold = 2}); - - int _consecutive = 0; - bool historyEmpty = false; - - bool recordOffload({required bool bankedRecords}) { - if (bankedRecords) { - _consecutive = 0; - historyEmpty = false; - return false; - } - _consecutive++; - if (!historyEmpty && _consecutive >= quietThreshold) { - historyEmpty = true; - return true; - } - return false; - } - - void reset() { - _consecutive = 0; - historyEmpty = false; - } -} - // ── detector 5: stuck strap ────────────────────────────────────────────────── /// The strap reports newer data than us but our persisted frontier hasn't moved /// for ≥10 min while the strap is >5 min ahead ⇒ stuck; action = defensive @@ -440,3 +487,59 @@ class StuckStrapDetector { _lastAdvanceWall = null; } } + +// ── meta-layer: staleness escalation ───────────────────────────────────────── +/// Every mechanism above (reconnect backoff, keep-alive, periodic backfill, +/// the OS-level wake sources on both platforms) can fail in SEQUENCE with +/// nothing noticing — an aggressive OEM battery killer, a lost race between +/// background wake sources, a band left out of BLE range for days. This is +/// the top-of-the-ladder backstop: watches how long it's been since we last +/// durably banked a record (the `rec_ts_hw` cursor — the same "honest +/// frontier" RecordGate/backfill policies already trust) and escalates in +/// tiers, ultimately driving the one universal backstop on both platforms — +/// a human reopening the app. +enum StalenessTier { + /// Recently synced — nothing to show. + fresh, + + /// Quiet in-app signal only (no OS notification) — long enough to be + /// worth a subtle indicator, not long enough to interrupt the user. + quiet, + + /// Long enough that a silent failure is more likely than "just hasn't + /// had new data" — worth an OS notification asking the user to reopen + /// the app (the only thing that can restart every wake mechanism at once). + notify, +} + +const int kStalenessQuietSeconds = 12 * 3600; // 12h +const int kStalenessNotifySeconds = 48 * 3600; // 48h + +/// Tier for [secondsSinceLastRecord] (wall-now minus the `rec_ts_hw` cursor). +/// Pure threshold lookup — callers own actually presenting the signal/ +/// notification and any de-duplication/cooldown around repeated firing. +StalenessTier stalenessTierFor(int secondsSinceLastRecord) { + if (secondsSinceLastRecord >= kStalenessNotifySeconds) { + return StalenessTier.notify; + } + if (secondsSinceLastRecord >= kStalenessQuietSeconds) { + return StalenessTier.quiet; + } + return StalenessTier.fresh; +} + +/// Whether enough time has passed since the last staleness OS notification to +/// fire another one. Distinct from [stalenessTierFor]'s threshold: a band +/// that stays lost for a week shouldn't get a notification every single +/// background cycle (which can run every ~15 min) — but SHOULD get reminded +/// periodically rather than only once, since a single missed/dismissed +/// notification shouldn't be the only chance to recover. [renotifyAfter] +/// defaults to the same width as the notify threshold itself. +bool shouldRenotifyStaleness( + DateTime? lastNotifiedAt, + DateTime now, { + Duration renotifyAfter = const Duration(seconds: kStalenessNotifySeconds), +}) { + if (lastNotifiedAt == null) return true; + return now.difference(lastNotifiedAt) >= renotifyAfter; +} diff --git a/lib/ui/profile/profile_screen.dart b/lib/ui/profile/profile_screen.dart index f204340f..df5323d1 100644 --- a/lib/ui/profile/profile_screen.dart +++ b/lib/ui/profile/profile_screen.dart @@ -75,6 +75,14 @@ class ProfileScreen extends StatelessWidget { // Android: battery-optimization (Doze) exemption — without it the OS // can freeze the background BLE session between events overnight. if (Platform.isAndroid) const _KeepAliveRow(), + // iOS: force-quitting the app is an unrecoverable OS contract — no + // background relaunch of any kind (CoreBluetooth restoration, + // BGTaskScheduler) fires again until the user manually reopens it + // once. Not fixable in code; this is the one-time explanation so + // "why did my data stop updating" has an answer, tied to the + // staleness notification (checkSyncStaleness) that eventually + // prompts the user to do exactly that. + if (Platform.isIOS) const _ForceQuitInfoRow(), const SizedBox(height: Sp.x6), @@ -1085,11 +1093,18 @@ class _KeepAliveRow extends StatefulWidget { class _KeepAliveRowState extends State<_KeepAliveRow> { bool? _exempt; // null = still checking + // Some OEMs (Xiaomi/Huawei/Honor/Oppo/Vivo/OnePlus) gate background survival + // behind a SEPARATE autostart/protected-apps allowlist the stock battery + // exemption above doesn't cover — see AndroidBackground.needsOemAutostartSettings. + // There's no OS-queryable "is it allowed" state for this (unlike the battery + // exemption), so this row is a one-shot action, not a toggle. + bool _showOemRow = false; @override void initState() { super.initState(); _refresh(); + _checkOem(); } Future _refresh() async { @@ -1097,6 +1112,11 @@ class _KeepAliveRowState extends State<_KeepAliveRow> { if (mounted) setState(() => _exempt = v); } + Future _checkOem() async { + final v = await context.read().needsOemAutostartSettings(); + if (mounted) setState(() => _showOemRow = v); + } + @override Widget build(BuildContext context) { return Column( @@ -1123,6 +1143,59 @@ class _KeepAliveRowState extends State<_KeepAliveRow> { await _refresh(); }, ), + if (_showOemRow) + ListRow( + icon: OsIcon.battery, + title: 'Allow auto-start', + subtitle: 'Your phone maker needs one more step to keep ' + 'syncing overnight — tap to open it', + onTap: () => context.read().openOemAutostartSettings(), + ), + ]), + ], + ); + } +} + +// ── iOS "force-quit" explainer ────────────────────────────────────────────── +// Swiping OpenStrap away in the app switcher is an unrecoverable OS contract: +// it explicitly opts the app out of ALL background relaunch — CoreBluetooth +// restoration, BGProcessingTask, BGAppRefreshTask — until the user manually +// reopens it once. There's no code fix for this; the best we can do is make +// sure the user understands why, since the eventual staleness notification +// (see sync/background_sync.dart's checkSyncStaleness) tells them WHAT to do +// ("reopen OpenStrap") but not WHY it stopped in the first place. +class _ForceQuitInfoRow extends StatelessWidget { + const _ForceQuitInfoRow(); + + @override + Widget build(BuildContext context) { + return Column( + crossAxisAlignment: CrossAxisAlignment.start, + children: [ + const SizedBox(height: Sp.x3), + _SettingsCard(rows: [ + ListRow( + icon: OsIcon.battery, + title: 'Force-quitting pauses background sync', + subtitle: 'If you swipe OpenStrap away, reopen it once to resume', + onTap: () => showInfoSheet( + context, + title: 'Force-quitting pauses background sync', + body: 'Swiping OpenStrap away in the app switcher tells iOS to ' + 'stop it completely — including background sync. This is ' + 'an iOS rule that applies to every app, not something ' + 'OpenStrap can change.', + bullets: const [ + 'Your data is never lost — the band keeps recording, and ' + 'the next sync catches everything up.', + 'To resume background syncing, just open OpenStrap once — ' + 'you don\'t need to do anything else.', + 'If you get a "hasn\'t synced in a while" notification, ' + 'this is usually why.', + ], + ), + ), ]), ], ); diff --git a/test/ble_state_test.dart b/test/ble_state_test.dart index 26b34bde..e2021566 100644 --- a/test/ble_state_test.dart +++ b/test/ble_state_test.dart @@ -225,6 +225,64 @@ void main() { }); }); + group('CounterRegressionDetector (band-reboot signal)', () { + test('ascending counters never trip', () { + final d = CounterRegressionDetector(); + expect(d.feed(1), isFalse); + expect(d.feed(100), isFalse); + expect(d.feed(101), isFalse); + expect(d.regressions, 0); + }); + + test('equal counters (duplicate/retried frame) do not trip', () { + final d = CounterRegressionDetector(); + expect(d.feed(50), isFalse); + expect(d.feed(50), isFalse); + expect(d.regressions, 0); + }); + + test('a real drop (band reboot) trips and is counted', () { + final d = CounterRegressionDetector(); + d.feed(5000); + expect(d.feed(3), isTrue); + expect(d.regressions, 1); + // Keeps counting further regressions (not one-shot — every reboot matters). + d.feed(10); + expect(d.feed(4), isTrue); + expect(d.regressions, 2); + }); + + test('u32 wraparound near the top of the range is not a regression', () { + final d = CounterRegressionDetector(); + d.feed(0xFFFFFFFF - 500000); + expect(d.feed(200), isFalse); // wrapped, not rebooted + expect(d.regressions, 0); + }); + + test('seeding from the durable counter_hw cursor catches a regression ' + 'across a reconnect', () { + final d = CounterRegressionDetector(seedCounter: 9000); + expect(d.feed(12), isTrue); // first record after reconnect already low + expect(d.regressions, 1); + }); + + test('first feed after construction with no seed never trips', () { + final d = CounterRegressionDetector(); + expect(d.feed(0), isFalse); + expect(d.regressions, 0); + }); + + test('reseed replaces the last-seen counter without resetting the ' + 'lifetime regression count', () { + final d = CounterRegressionDetector(seedCounter: 100); + d.feed(10); // regression #1 + expect(d.regressions, 1); + d.reseed(500); + expect(d.feed(600), isFalse); // ascending from the new seed + expect(d.regressions, 1); // lifetime count untouched by reseed + }); + }); + group('AckRetryPolicy (verified batch-ACK writes)', () { const p = AckRetryPolicy( maxAttempts: 3, @@ -247,6 +305,54 @@ void main() { }); }); + group('ChunkFailureLedger (persistent-per-token ACK failure tracking)', () { + test('a token that never fails never quarantines', () { + final l = ChunkFailureLedger(); + expect(l.failureCount('tok-a'), 0); + expect(l.shouldQuarantine('tok-a'), isFalse); + }); + + test('failures accumulate per-token, independent of other tokens', () { + final l = ChunkFailureLedger(quarantineThreshold: 3); + expect(l.recordFailure('tok-a'), 1); + expect(l.recordFailure('tok-a'), 2); + expect(l.recordFailure('tok-b'), 1); // independent counter + expect(l.failureCount('tok-a'), 2); + expect(l.failureCount('tok-b'), 1); + expect(l.shouldQuarantine('tok-a'), isFalse); + }); + + test('crosses the quarantine threshold on the Nth consecutive failure', + () { + final l = ChunkFailureLedger(quarantineThreshold: 3); + l.recordFailure('tok-a'); + l.recordFailure('tok-a'); + expect(l.shouldQuarantine('tok-a'), isFalse); + l.recordFailure('tok-a'); + expect(l.shouldQuarantine('tok-a'), isTrue); + }); + + test('a success clears tracking for that token only', () { + final l = ChunkFailureLedger(quarantineThreshold: 2); + l.recordFailure('tok-a'); + l.recordFailure('tok-a'); + l.recordFailure('tok-b'); + expect(l.shouldQuarantine('tok-a'), isTrue); + l.recordSuccess('tok-a'); + expect(l.failureCount('tok-a'), 0); + expect(l.shouldQuarantine('tok-a'), isFalse); + // tok-b untouched by tok-a's success. + expect(l.failureCount('tok-b'), 1); + }); + + test('a token can re-accumulate failures after a prior success', () { + final l = ChunkFailureLedger(quarantineThreshold: 2); + l.recordFailure('tok-a'); + l.recordSuccess('tok-a'); + expect(l.recordFailure('tok-a'), 1); // starts fresh, not from 1 again + }); + }); + group('DeriveDebouncer coalesce logic', () { const d = DeriveDebouncer( staleQuietPeriod: Duration(seconds: 12), diff --git a/test/db_integrity_test.dart b/test/db_integrity_test.dart index 99acf62f..392b5968 100644 --- a/test/db_integrity_test.dart +++ b/test/db_integrity_test.dart @@ -176,4 +176,142 @@ void main() { // New day → imported. expect(await payload('2026-06-30'), '{"src":"foreign"}'); }); + + group('sync_ledger real per-chunk rows + sync_quarantine reader', () { + test('distinct chunk_ids do not collide (the old "capture"-only bug)', + () async { + // Previously every call site defaulted to chunk_id='capture', so two + // different historical batches would overwrite the same row instead of + // each getting their own history. + await LocalDb.upsertSyncLedgerEntry( + chunkId: 'batch:aa', + kind: 'historical_batch', + status: 'acked', + metaPatch: {'records': 10}, + ); + await LocalDb.upsertSyncLedgerEntry( + chunkId: 'batch:bb', + kind: 'historical_batch', + status: 'ack_failed', + lastError: 'ack_write_exhausted', + metaPatch: {'ack_failures': 2}, + ); + + final a = await LocalDb.syncLedgerEntry('batch:aa'); + final b = await LocalDb.syncLedgerEntry('batch:bb'); + expect(a, isNotNull); + expect(b, isNotNull); + expect(a!['status'], 'acked'); + expect(b!['status'], 'ack_failed'); + + final all = await LocalDb.syncLedger(); + expect( + all.where((r) => r['chunk_id'] == 'batch:aa' + || r['chunk_id'] == 'batch:bb').length, + 2, + ); + }); + + test('a chunk_id survives repeated updates (persistent failure trail)', + () async { + await LocalDb.upsertSyncLedgerEntry( + chunkId: 'batch:cc', + kind: 'historical_batch', + status: 'ack_failed', + metaPatch: {'ack_failures': 1}, + ); + await LocalDb.upsertSyncLedgerEntry( + chunkId: 'batch:cc', + kind: 'historical_batch', + status: 'ack_failed', + metaPatch: {'ack_failures': 2}, + ); + await LocalDb.upsertSyncLedgerEntry( + chunkId: 'batch:cc', + kind: 'historical_batch', + status: 'ack_failed', + metaPatch: {'ack_failures': 3}, + ); + + final row = await LocalDb.syncLedgerEntry('batch:cc'); + expect(row, isNotNull); + // meta_json patches merge (shallow), so the latest failure count wins + // while created_at is preserved across the updates. + expect(row!['status'], 'ack_failed'); + final meta = row['meta_json'] as String; + expect(meta.contains('"ack_failures":3'), isTrue); + }); + + test('quarantined chunks are retrievable (previously write-only)', + () async { + await LocalDb.quarantineSyncChunk( + kind: 'historical_batch', + payloadJson: '{"token":"deadbeef","ack_failures":3}', + reason: 'persistent_ack_failure', + ); + final quarantined = await LocalDb.quarantinedSyncChunks(); + expect(quarantined, isNotEmpty); + expect( + quarantined.any((r) => r['reason'] == 'persistent_ack_failure'), + isTrue, + ); + }); + }); + + group('commitSyncBatch onCheckpoint (per-phase diagnostic logging)', () { + test('fires all three checkpoints in order on a normal commit', () async { + const ts = 1790000000; + final messages = []; + await LocalDb.commitSyncBatch( + [_raw(ts, 9001)], + [_sample(ts, 9001, [800])], + trimToken: 'deadbeef', + onCheckpoint: messages.add, + ); + expect(messages.length, 3); + expect(messages[0], startsWith('decoded_archive_queued')); + expect(messages[1], 'decoded_archive_committed'); + expect(messages[2], startsWith('cursor_advanced')); + expect(messages[2], contains('trim=true')); + + // The commit itself actually happened — checkpoints are observability, + // not a gate. + final db = await LocalDb.instance; + final rows = + await db.query('decoded_onehz', where: 'counter = ?', whereArgs: [9001]); + expect(rows, isNotEmpty); + }); + + test('a throwing onCheckpoint callback never aborts the commit', () async { + const ts = 1790000100; + var calls = 0; + await LocalDb.commitSyncBatch( + [_raw(ts, 9002)], + [_sample(ts, 9002, [810])], + onCheckpoint: (_) { + calls++; + throw StateError('a misbehaving logger must not break the commit'); + }, + ); + expect(calls, 3); // still fired at every phase despite always throwing + + final db = await LocalDb.instance; + final rows = + await db.query('decoded_onehz', where: 'counter = ?', whereArgs: [9002]); + expect(rows, isNotEmpty); // the commit succeeded regardless + }); + + test('with no onCheckpoint given, nothing is called and commit still works', + () async { + const ts = 1790000200; + await LocalDb.commitSyncBatch( + [_raw(ts, 9003)], + [_sample(ts, 9003, [820])], + ); + final db = await LocalDb.instance; + final rows = + await db.query('decoded_onehz', where: 'counter = ?', whereArgs: [9003]); + expect(rows, isNotEmpty); + }); + }); } diff --git a/test/derive_concurrency_test.dart b/test/derive_concurrency_test.dart new file mode 100644 index 00000000..79331865 --- /dev/null +++ b/test/derive_concurrency_test.dart @@ -0,0 +1,117 @@ +// runWithConcurrency: the bounded worker-pool primitive that replaced the +// fully-sequential per-day `for` loops in DerivationEngine.run()/runDays()/ +// rescanRecent(). A multi-day backlog sweep used to process one day fully +// (isolate spawns + compute) before starting the next, leaving every core +// but one idle. This pool lets several days' isolate work run genuinely +// concurrently across cores, with a continuous work-queue (not fixed +// batches) so a mix of fast/slow days keeps every lane busy. + +import 'dart:async'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:openstrap_edge/compute/derivation_engine.dart'; + +void main() { + group('runWithConcurrency', () { + test('empty items resolves immediately, worker never called', () async { + var calls = 0; + await runWithConcurrency(const [], 3, (item) async { + calls++; + }); + expect(calls, 0); + }); + + test('every item is processed exactly once, regardless of concurrency', + () async { + final items = List.generate(10, (i) => i); + final seen = []; + await runWithConcurrency(items, 3, (item) async { + seen.add(item); + }); + seen.sort(); + expect(seen, items); + }); + + test('concurrency=1 behaves like a plain sequential loop', () async { + final items = [1, 2, 3, 4]; + final completionOrder = []; + await runWithConcurrency(items, 1, (item) async { + // Even with an artificial delay, concurrency=1 means strict + // one-at-a-time — completion order must match input order exactly. + await Future.delayed(Duration.zero); + completionOrder.add(item); + }); + expect(completionOrder, items); + }); + + test('a concurrency higher than the item count is clamped harmlessly', + () async { + final items = [1, 2]; + var calls = 0; + await runWithConcurrency(items, 50, (item) async { + calls++; + }); + expect(calls, 2); // never spawns more lanes than items + }); + + test( + 'the first `concurrency` items start WITHOUT waiting for earlier ones ' + 'to finish — real parallelism, not fixed lock-step batches', () async { + // 3 items, concurrency=3, each holds until released — if they were + // sequential, item 1 would never even start until item 0's gate opens. + // With true concurrency, all 3 start immediately. + final gates = List.generate(3, (_) => Completer()); + final started = []; + final done = runWithConcurrency( + [0, 1, 2], + 3, + (item) async { + started.add(item); + await gates[item].future; + }, + ); + // Give the event loop a beat to let all 3 lanes actually start. + await Future.delayed(Duration.zero); + expect(started.toSet(), {0, 1, 2}); // all three in flight simultaneously + for (final g in gates) { + g.complete(); + } + await done; + }); + + test( + 'a slow first item does NOT block a fast later item from finishing ' + 'first (continuous queue, not fixed batches)', () async { + final finishOrder = []; + final slowGate = Completer(); + final done = runWithConcurrency( + [0, 1], + 2, + (item) async { + if (item == 0) { + await slowGate.future; // item 0 is slow + } + finishOrder.add(item); + }, + ); + // Item 1 should complete well before item 0's gate ever opens. + await Future.delayed(const Duration(milliseconds: 20)); + expect(finishOrder, [1]); + slowGate.complete(); + await done; + expect(finishOrder, [1, 0]); + }); + + test('a free lane immediately picks up the next queued item', () async { + // concurrency=1 over 3 items where each takes a beat — the single lane + // must move on to the next item as soon as the current one resolves, + // without any gap requiring external re-triggering. + final order = []; + await runWithConcurrency([10, 20, 30], 1, (item) async { + await Future.delayed(const Duration(milliseconds: 5)); + order.add(item); + }); + expect(order, [10, 20, 30]); + }); + }); +} diff --git a/test/headless_gate_test.dart b/test/headless_gate_test.dart new file mode 100644 index 00000000..2265659e --- /dev/null +++ b/test/headless_gate_test.dart @@ -0,0 +1,97 @@ +// HeadlessSyncGate: mutual exclusion across the three iOS headless wake +// sources (BLE-restore, BGProcessingTask, BGAppRefreshTask) + the skip-streak +// telemetry that makes repeated wake-source collisions observable instead of +// a single easy-to-miss debugPrint line. + +import 'dart:async'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:openstrap_edge/sync/headless_gate.dart'; + +void main() { + setUp(() => HeadlessSyncGate.resetForTest()); + + test('a solo run is never a skip and leaves no streak', () async { + final result = await HeadlessSyncGate.tryRun('owner_a', () async => 1); + expect(result, 1); + expect(HeadlessSyncGate.consecutiveSkipsFor('owner_a'), 0); + expect(HeadlessSyncGate.totalSkips, 0); + }); + + test('a collision skips the second caller and returns null', () async { + final gateHeld = Completer(); + final releaseGate = Completer(); + final firstRun = HeadlessSyncGate.tryRun('owner_a', () async { + gateHeld.complete(); + await releaseGate.future; + }); + await gateHeld.future; + + final second = await HeadlessSyncGate.tryRun('owner_b', () async => 2); + expect(second, isNull); + expect(HeadlessSyncGate.consecutiveSkipsFor('owner_b'), 1); + expect(HeadlessSyncGate.totalSkips, 1); + + releaseGate.complete(); + await firstRun; + }); + + test('consecutive skips accumulate per-owner independently', () async { + final gateHeld = Completer(); + final releaseGate = Completer(); + final firstRun = HeadlessSyncGate.tryRun('owner_a', () async { + gateHeld.complete(); + await releaseGate.future; + }); + await gateHeld.future; + + await HeadlessSyncGate.tryRun('owner_b', () async => 2); // skip #1 + await HeadlessSyncGate.tryRun('owner_b', () async => 2); // skip #2 + await HeadlessSyncGate.tryRun('owner_c', () async => 3); // owner_c skip #1 + + expect(HeadlessSyncGate.consecutiveSkipsFor('owner_b'), 2); + expect(HeadlessSyncGate.consecutiveSkipsFor('owner_c'), 1); + expect(HeadlessSyncGate.totalSkips, 3); + + releaseGate.complete(); + await firstRun; + }); + + test('a successful run resets that owner\'s own streak, not others\'', + () async { + final gateHeld = Completer(); + final releaseGate = Completer(); + final firstRun = HeadlessSyncGate.tryRun('owner_a', () async { + gateHeld.complete(); + await releaseGate.future; + }); + await gateHeld.future; + await HeadlessSyncGate.tryRun('owner_b', () async => 2); // skip + await HeadlessSyncGate.tryRun('owner_c', () async => 3); // skip + releaseGate.complete(); + await firstRun; + + expect(HeadlessSyncGate.consecutiveSkipsFor('owner_b'), 1); + expect(HeadlessSyncGate.consecutiveSkipsFor('owner_c'), 1); + + // owner_b finally gets to run — its OWN streak clears; owner_c's doesn't. + await HeadlessSyncGate.tryRun('owner_b', () async => 4); + expect(HeadlessSyncGate.consecutiveSkipsFor('owner_b'), 0); + expect(HeadlessSyncGate.consecutiveSkipsFor('owner_c'), 1); + }); + + test('busy reflects gate ownership across the run', () async { + expect(HeadlessSyncGate.busy, isFalse); + final gateHeld = Completer(); + final releaseGate = Completer(); + final run = HeadlessSyncGate.tryRun('owner_a', () async { + gateHeld.complete(); + await releaseGate.future; + }); + await gateHeld.future; + expect(HeadlessSyncGate.busy, isTrue); + releaseGate.complete(); + await run; + expect(HeadlessSyncGate.busy, isFalse); + }); +} diff --git a/test/sync_policy_test.dart b/test/sync_policy_test.dart index fea8216d..b78e597b 100644 --- a/test/sync_policy_test.dart +++ b/test/sync_policy_test.dart @@ -279,6 +279,85 @@ void main() { }); }); + group('FrameCorruptionDetector', () { + test('does not trip below minSamples even at 100% invalid', () { + final d = FrameCorruptionDetector(minSamples: 20); + for (var i = 0; i < 19; i++) { + expect(d.feed(false), isFalse); + } + expect(d.tripped, isFalse); + }); + + test('trips once the window crosses the corruption-rate threshold', () { + final d = FrameCorruptionDetector( + windowSize: 50, + rateThreshold: 0.2, + minSamples: 20, + ); + // 20 valid frames — enough samples, 0% corrupt, must not trip. + for (var i = 0; i < 20; i++) { + expect(d.feed(true), isFalse); + } + expect(d.tripped, isFalse); + // Now push the rate over 20% within the window. + bool trippedNow = false; + for (var i = 0; i < 10; i++) { + if (d.feed(false)) trippedNow = true; + } + expect(trippedNow, isTrue); + expect(d.tripped, isTrue); + }); + + test('one-shot — does not re-report after tripping', () { + final d = FrameCorruptionDetector( + windowSize: 10, + rateThreshold: 0.2, + minSamples: 5, + ); + for (var i = 0; i < 5; i++) { + d.feed(false); + } + // First feed past minSamples at 100% invalid trips. + var tripCount = 0; + for (var i = 0; i < 5; i++) { + if (d.feed(false)) tripCount++; + } + expect(tripCount, lessThanOrEqualTo(1)); + expect(d.tripped, isTrue); + }); + + test('a healthy link (occasional blip under threshold) never trips', () { + final d = FrameCorruptionDetector( + windowSize: 50, + rateThreshold: 0.2, + minSamples: 20, + ); + // 100 frames, ~5% corrupt — well under the 20% threshold. + for (var i = 0; i < 100; i++) { + final valid = i % 20 != 0; // 1 in 20 invalid = 5% + expect(d.feed(valid), isFalse); + } + expect(d.tripped, isFalse); + }); + + test('reset clears the window and tripped state', () { + final d = FrameCorruptionDetector( + windowSize: 10, + rateThreshold: 0.2, + minSamples: 5, + ); + for (var i = 0; i < 10; i++) { + d.feed(false); + } + expect(d.tripped, isTrue); + d.reset(); + expect(d.tripped, isFalse); + for (var i = 0; i < 4; i++) { + expect(d.feed(false), isFalse); // below minSamples again + } + }); + }); + group('PostBondTimeoutLoopDetector', () { test('trips after 2 bond→quick(<=8s)-timeouts', () { final d = PostBondTimeoutLoopDetector(); @@ -470,4 +549,76 @@ void main() { expect(kLinkFreshnessSeconds, lessThan(kLivenessFuseSeconds)); }); }); + + group('isCorruptFutureRtc (GET_DATA_RANGE sanity gate)', () { + const wall = 1750000000; + + test('a plausible newest timestamp is not corrupt', () { + expect(isCorruptFutureRtc(wall - 3600, wall), isFalse); + expect(isCorruptFutureRtc(wall, wall), isFalse); + }); + + test('exactly at the future margin is not corrupt', () { + expect(isCorruptFutureRtc(wall + kFutureMargin, wall), isFalse); + }); + + test('past the future margin is flagged corrupt', () { + expect(isCorruptFutureRtc(wall + kFutureMargin + 1, wall), isTrue); + expect(isCorruptFutureRtc(wall + 365 * 86400, wall), isTrue); // a year out + }); + }); + + group('stalenessTierFor (meta-layer: staleness escalation)', () { + test('recently synced is fresh', () { + expect(stalenessTierFor(0), StalenessTier.fresh); + expect(stalenessTierFor(3600), StalenessTier.fresh); + expect(stalenessTierFor(kStalenessQuietSeconds - 1), StalenessTier.fresh); + }); + + test('12h-48h is the quiet in-app tier', () { + expect(stalenessTierFor(kStalenessQuietSeconds), StalenessTier.quiet); + expect(stalenessTierFor(24 * 3600), StalenessTier.quiet); + expect( + stalenessTierFor(kStalenessNotifySeconds - 1), + StalenessTier.quiet, + ); + }); + + test('48h+ escalates to an OS notification', () { + expect(stalenessTierFor(kStalenessNotifySeconds), StalenessTier.notify); + expect(stalenessTierFor(7 * 86400), StalenessTier.notify); // a week gone + }); + }); + + group('shouldRenotifyStaleness (re-fire cooldown)', () { + final now = DateTime(2026, 1, 10, 12); + + test('never notified before → always fires', () { + expect(shouldRenotifyStaleness(null, now), isTrue); + }); + + test('within the cooldown window → does not re-fire', () { + final last = now.subtract(const Duration(hours: 1)); + expect(shouldRenotifyStaleness(last, now), isFalse); + }); + + test('past the cooldown window → fires again', () { + final last = now.subtract(const Duration(hours: 49)); + expect(shouldRenotifyStaleness(last, now), isTrue); + }); + + test('a custom cooldown is honoured', () { + final last = now.subtract(const Duration(hours: 2)); + expect( + shouldRenotifyStaleness(last, now, + renotifyAfter: const Duration(hours: 1)), + isTrue, + ); + expect( + shouldRenotifyStaleness(last, now, + renotifyAfter: const Duration(hours: 3)), + isFalse, + ); + }); + }); } From a102da1183cf137b237b06d98b5cb2d960840954 Mon Sep 17 00:00:00 2001 From: abdulsaheel Date: Wed, 8 Jul 2026 09:02:19 +0530 Subject: [PATCH 2/2] ci: strip Watch companion from unsigned IPA; re-sign ad-hoc native-assets frameworks The Watch app's Info.plist carries a literal WKCompanionAppBundleIdentifier that must exactly match the phone app's bundle id. Every free-account resigner (Sideloadly, AltStore) mints its own unique per-user bundle id without patching that unrelated nested plist key, so the Watch companion fails install for every sideloaded user under every resigner. Drop it from the unsigned CI artifact only; source builds via Xcode/flutter run keep full Watch support with a consistent signing identity. Also adds a "Re-sign ad-hoc native-assets frameworks" build phase: Dart's native-assets pipeline copies compiled frameworks into the app bundle via Flutter's own script rather than Xcode's auto-signing Copy Files mechanism, so they can end up ad-hoc-signed instead of matching the app's real signing identity, which fails code-signing validation on-device. Runs after both Embed Frameworks and CP Embed Pods Frameworks, re-signs only frameworks actually detected as ad-hoc via codesign -dv, and is a verified no-op when code signing isn't active (confirmed in a local --no-codesign build log). --- .github/workflows/build.yml | 27 +++++++++++++++++++++++++++ ios/Runner.xcodeproj/project.pbxproj | 17 +++++++++++++++++ 2 files changed, 44 insertions(+) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index ad773cf1..bf0fc1ac 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -77,6 +77,23 @@ jobs: # Note: free-account sideloads re-sign with a different team id, so App Group # features (home widget + Live Activity) may not work; there is no iOS OTA # auto-update either (that path installs an .apk). Attached to the same release. + # + # The Watch companion app is deliberately NOT embedded in this artifact — see + # the "Strip Watch companion" step below. It is structurally impossible for + # it to survive a per-user free-account resign: the Watch app's Info.plist + # carries a literal WKCompanionAppBundleIdentifier string that must exactly + # equal the phone app's bundle id, and every free-account resign (Sideloadly, + # AltStore, ...) mints its own unique per-user bundle id suffix to satisfy + # Apple's global App-ID-uniqueness rule but has no reason to know it also + # needs to patch that unrelated-looking key inside a different nested + # bundle's plist — none of them do. Unlike widget/Live-Activity extensions + # (which associate with the parent purely by PlugIns/ embedding + a simple + # bundle-id prefix, both of which survive naive resigning fine), this is not + # a Sideloadly bug to work around — it fails identically under every + # resigner. Full Watch support only ever works building from source with a + # consistent signing identity (see ios/Config/Signing.xcconfig + `flutter + # run`/Xcode direct install), where $(APP_BUNDLE_IDENTIFIER) threads through + # every target's plist, including this one, at build time. ios: runs-on: macos-latest permissions: @@ -104,6 +121,16 @@ jobs: - name: Build unsigned iOS app run: flutter build ios --release --no-codesign --dart-define-from-file=.env + # See the job-level comment above: the Watch companion cannot survive a + # per-user free-account resign under any sideloading tool (Sideloadly, + # AltStore, ...) — it fails install for every such user, every time, with + # no per-app-side fix possible, because the resign step would need to + # rewrite a cross-reference key inside a different nested bundle's plist + # that these tools don't special-case. Drop it here so the artifact this + # job ships actually installs; source builds keep full Watch support. + - name: Strip Watch companion (can't survive per-user resigning) + run: rm -rf "build/ios/iphoneos/Runner.app/Watch" + - name: Package unsigned IPA id: ipa run: | diff --git a/ios/Runner.xcodeproj/project.pbxproj b/ios/Runner.xcodeproj/project.pbxproj index 5300d6a1..7e20a7b7 100644 --- a/ios/Runner.xcodeproj/project.pbxproj +++ b/ios/Runner.xcodeproj/project.pbxproj @@ -379,6 +379,7 @@ 9705A1C41CF9048500538489 /* Embed Frameworks */, 3B06AD1E1E4923F5004D2608 /* Thin Binary */, 2DCD42F2BAD7B7B913EF5123 /* [CP] Embed Pods Frameworks */, + FADE0005FADE0005FADE0005 /* Re-sign ad-hoc native-assets frameworks */, ); buildRules = ( ); @@ -568,6 +569,22 @@ shellPath = /bin/sh; shellScript = "/bin/sh \"$FLUTTER_ROOT/packages/flutter_tools/bin/xcode_backend.sh\" build"; }; + FADE0005FADE0005FADE0005 /* Re-sign ad-hoc native-assets frameworks */ = { + isa = PBXShellScriptBuildPhase; + alwaysOutOfDate = 1; + buildActionMask = 2147483647; + files = ( + ); + inputPaths = ( + ); + name = "Re-sign ad-hoc native-assets frameworks"; + outputPaths = ( + ); + runOnlyForDeploymentPostprocessing = 0; + shellPath = /bin/sh; + shellScript = "if [ \"${CODE_SIGNING_ALLOWED}\" != \"YES\" ] || [ -z \"${EXPANDED_CODE_SIGN_IDENTITY}\" ]; then\n echo \"note: code signing not active for this build -- skipping native-assets framework re-sign.\"\n exit 0\nfi\nAPP_PATH=\"${TARGET_BUILD_DIR}/${WRAPPER_NAME}\"\nFRAMEWORKS_DIR=\"${APP_PATH}/Frameworks\"\nif [ -d \"${FRAMEWORKS_DIR}\" ]; then\n find \"${FRAMEWORKS_DIR}\" -maxdepth 1 -name \"*.framework\" | while IFS= read -r fw; do\n if codesign -dv \"${fw}\" 2>&1 | grep -q \"Signature=adhoc\"; then\n echo \"note: re-signing ad-hoc-signed framework: ${fw}\"\n codesign --force --sign \"${EXPANDED_CODE_SIGN_IDENTITY}\" --preserve-metadata=identifier,entitlements,flags \"${fw}\"\n fi\n done\nfi\n"; + showEnvVarsInLog = 0; + }; /* End PBXShellScriptBuildPhase section */ /* Begin PBXSourcesBuildPhase section */