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/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.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 */ 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, + ); + }); + }); }