- Chunk large HealthKit page persistence
- Browse files
-
bogdan committed a week ago
Showing 2 changed files
with
81 additions
and
31 deletions
+1
-1
HealthProbe/Doc/04-project/IMPLEMENTATION_STATUS.md
@@ -24,7 +24,7 @@ There are no real deployments, only test installations. Existing prototype datab |
||
| 24 | 24 |
| Area | Current Status | Target / Next Work | |
| 25 | 25 |
|------|----------------|--------------------| |
| 26 | 26 |
| Product docs | Updated | Keep `HealthProbe/Doc/README.md` as canonical index | |
| 27 |
-| HealthKit capture | Capture now opens one archive observation per user-visible snapshot, attaches HealthKit pages, deleted-object evidence, and type verification to that observation id before finishing it, no longer aborts initial full-history imports after a fixed 30-minute wall-clock cap while page-level HealthKit timeouts remain in place, and defers grouped observation summary/daily aggregate rebuilds until per-type verification instead of rebuilding after every imported page | Continue moving UI/cache reads to archive-backed observation ids and revisit adaptive paging/background collection separately | |
|
| 27 |
+| HealthKit capture | Capture now opens one archive observation per user-visible snapshot, attaches HealthKit pages, deleted-object evidence, and type verification to that observation id before finishing it, no longer aborts initial full-history imports after a fixed 30-minute wall-clock cap while page-level HealthKit timeouts remain in place, defers grouped observation summary/daily aggregate rebuilds until per-type verification instead of rebuilding after every imported page, and persists large HealthKit pages in smaller archive chunks while using a smaller anchored-query page size to avoid long monolithic SQLite stalls | Continue moving UI/cache reads to archive-backed observation ids and revisit fully adaptive paging/background collection separately | |
|
| 28 | 28 |
| SQLite archive | Archive v2 schema, snapshot-level observation grouping, differential write path, v2 verification/delete bookkeeping, daily aggregate rebuilds, integrity report, v2 record reads, SQL diff/count/aggregate/provenance/consolidation-evidence APIs, large synthetic diff pagination coverage, formal timing/memory metrics, and XCTest coverage are in place; the legacy `archive_samples` mirror has been removed | Continue moving capture/Dashboard actions to archive/cache DTOs | |
| 29 | 29 |
| Core Data cache | Initial programmatic Core Data model, full-cache rebuild service, read DTOs for observation/type/diff/health rows, and Dashboard archive-cache status wiring are in place | Move remaining export/report paths to cache DTOs and add targeted partial invalidation | |
| 30 | 30 |
| SwiftData cache | Exists; test builds now reset legacy prototype UI/archive/cache stores once for archive v2 so old SwiftData-only snapshots are not treated as backed-up observations. Metric timeout calibration, local device profile settings, operation logging, ContentView preview, Settings data maintenance, legacy detail/PDF views, unused legacy repair/observer services, Dashboard view/view-model access, and legacy anomaly/count-drop review have moved outside SwiftData or been removed. Remaining SwiftData imports are inventoried in [`SwiftData-Retirement-Inventory.md`](SwiftData-Retirement-Inventory.md) | Treat as disposable prototype data; stop returning/storing `HealthSnapshot` bridge handles before removing `ModelContainer` | |
+80
-30
HealthProbe/Services/HealthKitService.swift
@@ -1099,17 +1099,17 @@ final class HealthKitService {
|
||
| 1099 | 1099 |
anchor: anchor |
| 1100 | 1100 |
) |
| 1101 | 1101 |
} |
| 1102 |
- progress?.updateBlockProgress( |
|
| 1103 |
- typeIdentifier, |
|
| 1104 |
- detail: "Persisting delta page \(pageNumber)", |
|
| 1102 |
+ try await archivePage( |
|
| 1103 |
+ page, |
|
| 1104 |
+ sampleType: sampleType, |
|
| 1105 |
+ observationID: archiveObservationID, |
|
| 1106 |
+ progress: progress, |
|
| 1107 |
+ typeIdentifier: typeIdentifier, |
|
| 1108 |
+ progressDetail: "Persisting delta page \(pageNumber)", |
|
| 1105 | 1109 |
recordCount: previousDistribution.count, |
| 1106 |
- elapsedSeconds: Date().timeIntervalSince(progressStarted), |
|
| 1107 |
- samplesPerSecond: Self.samplesPerSecond( |
|
| 1108 |
- processedCount: processedEventCount, |
|
| 1109 |
- elapsedSeconds: Date().timeIntervalSince(progressStarted) |
|
| 1110 |
- ) |
|
| 1110 |
+ progressStarted: progressStarted, |
|
| 1111 |
+ processedEventCount: processedEventCount |
|
| 1111 | 1112 |
) |
| 1112 |
- try await archivePage(page, sampleType: sampleType, observationID: archiveObservationID) |
|
| 1113 | 1113 |
anchor = page.anchor |
| 1114 | 1114 |
|
| 1115 | 1115 |
if page.samples.isEmpty, page.deletedObjects.isEmpty, |
@@ -1196,17 +1196,17 @@ final class HealthKitService {
|
||
| 1196 | 1196 |
anchor: anchor |
| 1197 | 1197 |
) |
| 1198 | 1198 |
} |
| 1199 |
- progress?.updateBlockProgress( |
|
| 1200 |
- typeIdentifier, |
|
| 1201 |
- detail: "Persisting delta page \(pageNumber)", |
|
| 1199 |
+ try await archivePage( |
|
| 1200 |
+ page, |
|
| 1201 |
+ sampleType: sampleType, |
|
| 1202 |
+ observationID: archiveObservationID, |
|
| 1203 |
+ progress: progress, |
|
| 1204 |
+ typeIdentifier: typeIdentifier, |
|
| 1205 |
+ progressDetail: "Persisting delta page \(pageNumber)", |
|
| 1202 | 1206 |
recordCount: recordMap.count, |
| 1203 |
- elapsedSeconds: Date().timeIntervalSince(progressStarted), |
|
| 1204 |
- samplesPerSecond: Self.samplesPerSecond( |
|
| 1205 |
- processedCount: processedEventCount, |
|
| 1206 |
- elapsedSeconds: Date().timeIntervalSince(progressStarted) |
|
| 1207 |
- ) |
|
| 1207 |
+ progressStarted: progressStarted, |
|
| 1208 |
+ processedEventCount: processedEventCount |
|
| 1208 | 1209 |
) |
| 1209 |
- try await archivePage(page, sampleType: sampleType, observationID: archiveObservationID) |
|
| 1210 | 1210 |
anchor = page.anchor |
| 1211 | 1211 |
|
| 1212 | 1212 |
applyDistributionPage(page, sampleType: sampleType, to: &recordMap) |
@@ -1344,17 +1344,17 @@ final class HealthKitService {
|
||
| 1344 | 1344 |
anchor: anchor |
| 1345 | 1345 |
) |
| 1346 | 1346 |
} |
| 1347 |
- progress?.updateBlockProgress( |
|
| 1348 |
- typeIdentifier, |
|
| 1349 |
- detail: "Persisting import page \(pageNumber)", |
|
| 1347 |
+ try await archivePage( |
|
| 1348 |
+ page, |
|
| 1349 |
+ sampleType: sampleType, |
|
| 1350 |
+ observationID: archiveObservationID, |
|
| 1351 |
+ progress: progress, |
|
| 1352 |
+ typeIdentifier: typeIdentifier, |
|
| 1353 |
+ progressDetail: "Persisting import page \(pageNumber)", |
|
| 1350 | 1354 |
recordCount: recordCount, |
| 1351 |
- elapsedSeconds: Date().timeIntervalSince(progressStarted), |
|
| 1352 |
- samplesPerSecond: Self.samplesPerSecond( |
|
| 1353 |
- processedCount: processedEventCount, |
|
| 1354 |
- elapsedSeconds: Date().timeIntervalSince(progressStarted) |
|
| 1355 |
- ) |
|
| 1355 |
+ progressStarted: progressStarted, |
|
| 1356 |
+ processedEventCount: processedEventCount |
|
| 1356 | 1357 |
) |
| 1357 |
- try await archivePage(page, sampleType: sampleType, observationID: archiveObservationID) |
|
| 1358 | 1358 |
anchor = page.anchor |
| 1359 | 1359 |
|
| 1360 | 1360 |
for sample in page.samples {
|
@@ -1542,9 +1542,58 @@ final class HealthKitService {
|
||
| 1542 | 1542 |
} |
| 1543 | 1543 |
} |
| 1544 | 1544 |
|
| 1545 |
- private func archivePage(_ page: SampleDistributionPage, sampleType: HKSampleType, observationID: Int64) async throws {
|
|
| 1545 |
+ private func archivePage( |
|
| 1546 |
+ _ page: SampleDistributionPage, |
|
| 1547 |
+ sampleType: HKSampleType, |
|
| 1548 |
+ observationID: Int64, |
|
| 1549 |
+ progress: SnapshotFetchProgress? = nil, |
|
| 1550 |
+ typeIdentifier: String? = nil, |
|
| 1551 |
+ progressDetail: String? = nil, |
|
| 1552 |
+ recordCount: Int? = nil, |
|
| 1553 |
+ progressStarted: Date? = nil, |
|
| 1554 |
+ processedEventCount: Int? = nil |
|
| 1555 |
+ ) async throws {
|
|
| 1546 | 1556 |
let observedAt = Date() |
| 1547 |
- _ = try await archiveStore.upsertSamples(page.samples, observedAt: observedAt, observationID: observationID) |
|
| 1557 |
+ if !page.samples.isEmpty {
|
|
| 1558 |
+ let totalSampleBatches = max( |
|
| 1559 |
+ 1, |
|
| 1560 |
+ Int(ceil(Double(page.samples.count) / Double(DistributionCaptureConfiguration.archiveWriteChunkSize))) |
|
| 1561 |
+ ) |
|
| 1562 |
+ var batchNumber = 0 |
|
| 1563 |
+ var batchStart = 0 |
|
| 1564 |
+ while batchStart < page.samples.count {
|
|
| 1565 |
+ batchNumber += 1 |
|
| 1566 |
+ let batchEnd = min( |
|
| 1567 |
+ batchStart + DistributionCaptureConfiguration.archiveWriteChunkSize, |
|
| 1568 |
+ page.samples.count |
|
| 1569 |
+ ) |
|
| 1570 |
+ let sampleBatch = Array(page.samples[batchStart..<batchEnd]) |
|
| 1571 |
+ if let progress, let typeIdentifier, let progressDetail {
|
|
| 1572 |
+ let detail = totalSampleBatches == 1 |
|
| 1573 |
+ ? progressDetail |
|
| 1574 |
+ : "\(progressDetail) (\(batchNumber)/\(totalSampleBatches))" |
|
| 1575 |
+ progress.updateBlockProgress( |
|
| 1576 |
+ typeIdentifier, |
|
| 1577 |
+ detail: detail, |
|
| 1578 |
+ recordCount: recordCount, |
|
| 1579 |
+ elapsedSeconds: progressStarted.map { Date().timeIntervalSince($0) },
|
|
| 1580 |
+ samplesPerSecond: {
|
|
| 1581 |
+ guard let progressStarted, let processedEventCount else { return nil }
|
|
| 1582 |
+ return Self.samplesPerSecond( |
|
| 1583 |
+ processedCount: processedEventCount, |
|
| 1584 |
+ elapsedSeconds: Date().timeIntervalSince(progressStarted) |
|
| 1585 |
+ ) |
|
| 1586 |
+ }() |
|
| 1587 |
+ ) |
|
| 1588 |
+ } |
|
| 1589 |
+ _ = try await archiveStore.upsertSamples( |
|
| 1590 |
+ sampleBatch, |
|
| 1591 |
+ observedAt: observedAt, |
|
| 1592 |
+ observationID: observationID |
|
| 1593 |
+ ) |
|
| 1594 |
+ batchStart = batchEnd |
|
| 1595 |
+ } |
|
| 1596 |
+ } |
|
| 1548 | 1597 |
for deletedObject in page.deletedObjects {
|
| 1549 | 1598 |
try await archiveStore.recordDisappearance( |
| 1550 | 1599 |
sampleUUIDHash: HashService.sampleUUIDHash(deletedObject.uuid.uuidString), |
@@ -2390,6 +2439,7 @@ private extension SnapshotFetchProgress {
|
||
| 2390 | 2439 |
} |
| 2391 | 2440 |
|
| 2392 | 2441 |
private enum DistributionCaptureConfiguration {
|
| 2393 |
- static let queryPageLimit = 25_000 |
|
| 2442 |
+ static let queryPageLimit = 5_000 |
|
| 2394 | 2443 |
static let pageTimeoutSeconds: TimeInterval = 60 |
| 2444 |
+ static let archiveWriteChunkSize = 1_000 |
|
| 2395 | 2445 |
} |