diff --git a/src/js/app.js b/src/js/app.js index 9a7a2be..953dac1 100644 --- a/src/js/app.js +++ b/src/js/app.js @@ -259,14 +259,17 @@ class App { window.addEventListener('pagehide', pushOnHide); // Push immediately when app goes to background — a delayed timer may - // never fire on mobile (page freeze); pull on return to foreground + // never fire on mobile (page freeze); check for other devices' pushes + // on return to foreground and while visible document.addEventListener('visibilitychange', () => { if (document.visibilityState === 'hidden') { + syncManager.stopSnapshotWatch(); pushOnHide(); } else if (document.visibilityState === 'visible') { if (!syncManager.isAutoSyncAllowed('foreground')) return; - this.pullSyncedStateAndBlobs().catch((error) => { - Logger.debug('Sync: Auto-pull on foreground failed (non-critical):', error.message); + syncManager.startSnapshotWatch(); + syncManager.checkForNewSnapshot().catch((error) => { + Logger.debug('Sync: Check on foreground failed (non-critical):', error.message); }); } }); @@ -356,6 +359,8 @@ class App { // Stop background timers that survive disconnect otherwise: // sync auto-push and relay token re-registration (6h interval) syncManager.cancelAutoPush(); + syncManager.stopSnapshotWatch(); + syncManager.cancelPushConfirmation(); relayManager.stopReRegistrationTimer(); // Drop invites of the disconnecting account so they can't be // re-saved under the next account's storage @@ -589,6 +594,10 @@ class App { if (syncManager.isAutoSyncAllowed('start')) { runInitialSync(); } + if (document.visibilityState === 'visible' + && syncManager.isAutoSyncAllowed('foreground')) { + syncManager.startSnapshotWatch(); + } } } catch (dmError) { Logger.warn('Failed to init DM manager (non-critical):', dmError); diff --git a/src/js/config.js b/src/js/config.js index e266637..e6521ee 100644 --- a/src/js/config.js +++ b/src/js/config.js @@ -395,6 +395,7 @@ export const CONFIG = { syncDirty: (addr) => `pombo_sync_dirty_${addr.toLowerCase()}`, syncMode: (addr) => `pombo_sync_mode_${addr.toLowerCase()}`, syncAppliedTs: (addr) => `pombo_sync_applied_ts_${addr.toLowerCase()}`, + syncConfirmed: (addr) => `pombo_sync_confirmed_${addr.toLowerCase()}`, // Per-stream keys channelAccess: (streamId) => `pombo_channel_access_${streamId}` diff --git a/src/js/streamr.js b/src/js/streamr.js index 9a8add1..eb93cb0 100644 --- a/src/js/streamr.js +++ b/src/js/streamr.js @@ -2434,6 +2434,24 @@ class StreamrController { return message; } + /** + * Take the node out of a stream part it joined only to publish. A part + * with a live subscription is left alone: the subscription rides on it. + * @param {string} streamId + * @param {number} partition + */ + async leaveStreamPart(streamId, partition) { + if (!this.client) return; + if (this.subscriptions.get(streamId)?.[partition]) return; + try { + const node = this.client.getNode(); + await node.leave(`${streamId}#${partition}`); + Logger.debug(`Left ${streamId} p${partition}; stream parts joined: ${(await node.getStreamParts()).length}`); + } catch (error) { + Logger.debug(`leaveStreamPart ${streamId} p${partition} failed:`, error.message); + } + } + // ==================== KEYS STREAM (-4) ==================== /** diff --git a/src/js/syncManager.js b/src/js/syncManager.js index b2b7190..77da317 100644 --- a/src/js/syncManager.js +++ b/src/js/syncManager.js @@ -31,6 +31,27 @@ import { splitSyncPayload, reassembleSyncPayloads } from './syncChunks.js'; /** A snapshot is a RUN of messages, so the window must hold several of them. */ const SYNC_FETCH_COUNT = 60; +/** How often an open, visible app asks storage whether another device pushed. */ +const SYNC_CHECK_INTERVAL_MS = 60_000; + +/** Storage re-reads after a push, measured from the publish. */ +const SYNC_CONFIRM_AT_MS = [5_000, 10_000, 20_000, 40_000]; + +/** Rows beyond our own read back with them: pushes from other devices in between. */ +const SYNC_CONFIRM_SLACK = 10; + +/** Blob pushes are not read back; the overlay is kept this long for them to leave. */ +const SYNC_BLOB_LEAVE_AFTER_MS = 30_000; + +/** A confirmed state is sent again after this long: storage keeps rows only for the inbox retention. */ +const SYNC_CONFIRMED_MAX_AGE_MS = 7 * 24 * 60 * 60 * 1000; + +/** Row keys of this session's own pushes, so a check does not pull them back. */ +const SYNC_OWN_ROWS_KEPT = 64; + +/** Every sync message has its own throwaway publisher, so this names one row. */ +const rowKey = (row) => `${row.timestamp}:${String(row.publisherId || '').toLowerCase()}`; + const getNow = () => ( typeof performance !== 'undefined' && typeof performance.now === 'function' ? performance.now() @@ -70,6 +91,13 @@ class SyncManager { this.autoPushTimeout = null; this.autoPushRetryCount = 0; this.pushQueued = false; + this._pulledRowTs = 0; + this._ownRowKeys = new Set(); + this._publishing = false; + this._confirmRun = 0; + this._confirmTimer = null; + this._blobLeaveTimer = null; + this._snapshotWatch = null; } // ==================== Sync Mode (which unprompted triggers fire) ==================== @@ -182,6 +210,53 @@ class SyncManager { } } + // ==================== Confirmed state (what storage is known to hold) ==================== + + _confirmedKey() { + const address = authManager.getAddress(); + if (!address) return null; + const keyFn = CONFIG.storageKeys?.syncConfirmed; + return keyFn ? keyFn(address) : `pombo_sync_confirmed_${address.toLowerCase()}`; + } + + /** True when storage already holds this exact state from a recent push of ours. */ + _isConfirmedState(hash) { + const key = this._confirmedKey(); + if (!key) return false; + try { + const confirmed = JSON.parse(localStorage.getItem(key) || 'null'); + return confirmed?.hash === hash && Date.now() - confirmed.at < SYNC_CONFIRMED_MAX_AGE_MS; + } catch { + return false; + } + } + + _writeConfirmedState(hash) { + const key = this._confirmedKey(); + if (!key) return; + try { + localStorage.setItem(key, JSON.stringify({ hash, at: Date.now() })); + } catch { /* storage unavailable — the next push is just not skipped */ } + } + + async _stateHash(state) { + const digest = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(JSON.stringify(state))); + return Array.from(new Uint8Array(digest), (b) => b.toString(16).padStart(2, '0')).join(''); + } + + _notePulledRows(rows) { + for (const row of rows) { + if (row.timestamp > this._pulledRowTs) this._pulledRowTs = row.timestamp; + } + } + + _noteOwnRows(rows) { + for (const row of rows) this._ownRowKeys.add(rowKey(row)); + while (this._ownRowKeys.size > SYNC_OWN_ROWS_KEPT) { + this._ownRowKeys.delete(this._ownRowKeys.values().next().value); + } + } + /** * If a push was requested while another sync operation was running, * run it now that the operation finished (short debounce). @@ -373,6 +448,14 @@ class SyncManager { try { // Gather state from secureStorage const state = secureStorage.exportForSync(); + const hash = await this._stateHash(state); + if (this._isConfirmedState(hash)) { + Logger.info('Sync: State unchanged since the last confirmed push, not publishing'); + if (!this.autoPushTimeout && !this.pushQueued) { + this.clearDirty(); + } + return null; + } const payload = { type: 'sync', @@ -395,10 +478,20 @@ class SyncManager { channels: state.channels?.length ?? 0, messages: messages.length }); - for (const message of messages) { - await dmManager.sealAndPublish( - inboxStreamId, myAddress, message, STREAM_CONFIG.MESSAGE_STREAM.SYNC); + const rows = []; + this._publishing = true; + try { + for (const message of messages) { + const published = await dmManager.sealAndPublish( + inboxStreamId, myAddress, message, STREAM_CONFIG.MESSAGE_STREAM.SYNC); + const id = published?.messageId; + if (id) rows.push({ timestamp: id.timestamp, publisherId: id.publisherId }); + } + } finally { + this._publishing = false; } + this._noteOwnRows(rows); + this._confirmPush(inboxStreamId, hash, rows); this.lastSyncTs = payload.ts; Logger.info('Sync: Pushed state to storage nodes', { ts: payload.ts }); @@ -421,6 +514,106 @@ class SyncManager { } } + /** + * Read a push back from storage, then take the node out of the sync + * partition: the overlay is only needed to publish. An unconfirmed push + * leaves it all the same and marks the state dirty, so the next trigger + * sends it again. Rows from another device met on the way are pulled. + */ + _confirmPush(inboxStreamId, hash, rows) { + const run = ++this._confirmRun; + clearTimeout(this._confirmTimer); + const wanted = new Set(rows.map(rowKey)); + const startedAt = Date.now(); + const attempt = async (index) => { + if (run !== this._confirmRun) return; + let held = []; + try { + held = await streamrController.fetchPartitionHistory( + inboxStreamId, STREAM_CONFIG.MESSAGE_STREAM.SYNC, wanted.size + SYNC_CONFIRM_SLACK); + } catch { /* an unreachable node proves nothing */ } + if (run !== this._confirmRun) return; + const seen = new Set(held.map(rowKey)); + const confirmed = wanted.size > 0 && [...wanted].every((key) => seen.has(key)); + if (!confirmed && index + 1 < SYNC_CONFIRM_AT_MS.length) { + this._confirmTimer = setTimeout(() => attempt(index + 1), + Math.max(0, startedAt + SYNC_CONFIRM_AT_MS[index + 1] - Date.now())); + return; + } + if (confirmed) { + this._writeConfirmedState(hash); + Logger.info('Sync: Push confirmed by storage'); + } else if (wanted.size > 0) { + Logger.warn('Sync: Push not confirmed by storage, it will be sent again'); + this.markDirty(); + } + const foreign = held.some((row) => + !this._ownRowKeys.has(rowKey(row)) && row.timestamp > this._pulledRowTs); + // A push in flight re-joins the partition and leaves it when confirmed. + if (!this._publishing) { + await streamrController.leaveStreamPart(inboxStreamId, STREAM_CONFIG.MESSAGE_STREAM.SYNC); + } + if (foreign && this.isAutoSyncAllowed('foreground')) { + this._pullStateAndBlobs().catch((e) => + Logger.debug('Sync: Pull after confirm failed (non-critical):', e.message)); + } + }; + this._confirmTimer = setTimeout(() => attempt(0), SYNC_CONFIRM_AT_MS[0]); + } + + async _pullStateAndBlobs() { + const pulled = await this.pullSync(); + if (pulled !== null) { + await this.pullImageBlobs().catch((e) => + Logger.debug('Sync: Image blob pull failed (non-critical):', e.message)); + } + return pulled; + } + + /** + * Pull when storage holds a row from another device newer than anything + * read here. One `last: 1` read over HTTP, no overlay: the last row of a + * chunked push is its small manifest. + * @returns {Promise} whether a pull ran + */ + async checkForNewSnapshot() { + if (authManager.isGuestMode() || this.isSyncing) return false; + if (!this.isAutoSyncAllowed('foreground')) return false; + if (!await dmManager.hasInbox()) return false; + const inboxStreamId = this.getInboxStreamId(); + if (!inboxStreamId) return false; + const [newest] = await streamrController.fetchPartitionHistory( + inboxStreamId, STREAM_CONFIG.MESSAGE_STREAM.SYNC, 1); + if (!newest || this._ownRowKeys.has(rowKey(newest)) || newest.timestamp <= this._pulledRowTs) { + return false; + } + await this._pullStateAndBlobs(); + return true; + } + + /** Checks every SYNC_CHECK_INTERVAL_MS until stopped; run it only while the app is visible. */ + startSnapshotWatch() { + this.stopSnapshotWatch(); + this._snapshotWatch = setInterval(() => { + this.checkForNewSnapshot().catch((e) => + Logger.debug('Sync: Snapshot check failed (non-critical):', e.message)); + }, SYNC_CHECK_INTERVAL_MS); + } + + stopSnapshotWatch() { + clearInterval(this._snapshotWatch); + this._snapshotWatch = null; + } + + /** On disconnect: a confirmation still running would act on the next account. */ + cancelPushConfirmation() { + this._confirmRun++; + clearTimeout(this._confirmTimer); + clearTimeout(this._blobLeaveTimer); + this._pulledRowTs = 0; + this._ownRowKeys.clear(); + } + /** * Pull sync history and merge with local state. * Fetches all payloads, merges chronologically (older first, newer overwrites). @@ -473,6 +666,7 @@ class SyncManager { options.limit || SYNC_FETCH_COUNT ); const fetchCompletedAt = getNow(); + this._notePulledRows(messages); Logger.info('Sync: Fetched messages:', { count: messages.length }); @@ -573,6 +767,7 @@ class SyncManager { STREAM_CONFIG.MESSAGE_STREAM.SYNC, options.limit || SYNC_FETCH_COUNT ); + this._notePulledRows(retryMessages); const retryDecrypted = reassemble(await decryptBatch(retryMessages)); freshPayloads = retryDecrypted.filter(payload => !appliedTs.has(payload.ts)); } @@ -822,6 +1017,12 @@ class SyncManager { } Logger.debug(`Sync: pushImageBlobs done — scanned=${scanned} pushed=${count}`); + if (count > 0) { + clearTimeout(this._blobLeaveTimer); + this._blobLeaveTimer = setTimeout(() => { + streamrController.leaveStreamPart(inboxStreamId, STREAM_CONFIG.MESSAGE_STREAM.SYNC_BLOBS); + }, SYNC_BLOB_LEAVE_AFTER_MS); + } } /** diff --git a/tests/unit/streamr.core.test.js b/tests/unit/streamr.core.test.js index f2187d0..212fbb4 100644 --- a/tests/unit/streamr.core.test.js +++ b/tests/unit/streamr.core.test.js @@ -211,6 +211,21 @@ describe('StreamrController Core', () => { }); }); + // ==================== leaveStreamPart() ==================== + describe('leaveStreamPart()', () => { + it('leaves a part it only published to, and never a subscribed one', async () => { + const node = { leave: vi.fn().mockResolvedValue(undefined), getStreamParts: vi.fn().mockResolvedValue([]) }; + mockClient.getNode = vi.fn().mockReturnValue(node); + streamrController.subscriptions.set('0xme/Pombo-DM-1', { 0: {} }); + + await streamrController.leaveStreamPart('0xme/Pombo-DM-1', 0); + expect(node.leave).not.toHaveBeenCalled(); + + await streamrController.leaveStreamPart('0xme/Pombo-DM-1', 1); + expect(node.leave).toHaveBeenCalledWith('0xme/Pombo-DM-1#1'); + }); + }); + // ==================== publish() ==================== describe('publish()', () => { it('should throw if client is null', async () => { diff --git a/tests/unit/syncManager.extended.test.js b/tests/unit/syncManager.extended.test.js index 2892d0e..bbbe542 100644 --- a/tests/unit/syncManager.extended.test.js +++ b/tests/unit/syncManager.extended.test.js @@ -19,6 +19,7 @@ vi.mock('../../src/js/streamr.js', () => ({ getDMInboxId: vi.fn().mockReturnValue('0xabc/Pombo-DM-1'), publish: vi.fn().mockResolvedValue(undefined), fetchPartitionHistory: vi.fn().mockResolvedValue([]), + leaveStreamPart: vi.fn().mockResolvedValue(undefined), setDMPublishKey: vi.fn().mockResolvedValue(undefined), addDMDecryptKey: vi.fn().mockResolvedValue(undefined) }, @@ -95,7 +96,7 @@ vi.mock('../../src/js/identity.js', () => ({ } })); -import { syncManager } from '../../src/js/syncManager.js'; +import { syncManager, SYNC_MODES } from '../../src/js/syncManager.js'; import { authManager } from '../../src/js/auth.js'; import { dmManager } from '../../src/js/dm.js'; import { Logger } from '../../src/js/logger.js'; @@ -721,4 +722,105 @@ describe('syncManager extended', () => { expect(remoteMsg.text).toBe('hello'); }); }); + + describe('overlay only while publishing, periodic check over HTTP', () => { + const own = (timestamp) => ({ timestamp, publisherId: '0xeph' }); + const other = (timestamp) => ({ timestamp, publisherId: '0xother' }); + + beforeEach(() => { + vi.useFakeTimers(); + syncManager.cancelPushConfirmation(); + dmManager.sealAndPublish.mockResolvedValue({ messageId: { timestamp: 5000, publisherId: '0xEph' } }); + }); + + afterEach(() => { + syncManager.cancelPushConfirmation(); + syncManager.stopSnapshotWatch(); + vi.useRealTimers(); + }); + + it('reads a push back, leaves the sync partition, and does not publish the same state again', async () => { + await syncManager.pushSync(); + streamrController.fetchPartitionHistory.mockResolvedValue([own(5000)]); + await vi.advanceTimersByTimeAsync(5000); + + expect(streamrController.leaveStreamPart).toHaveBeenCalledWith('0xabc/Pombo-DM-1', 1); + dmManager.sealAndPublish.mockClear(); + await syncManager.pushSync(); + expect(dmManager.sealAndPublish).not.toHaveBeenCalled(); + }); + + it('leaves an unconfirmed push too, and sends that state again', async () => { + await syncManager.pushSync(); + streamrController.fetchPartitionHistory.mockResolvedValue([]); + await vi.advanceTimersByTimeAsync(40000); + + expect(streamrController.fetchPartitionHistory).toHaveBeenCalledTimes(4); + expect(streamrController.leaveStreamPart).toHaveBeenCalledTimes(1); + expect(syncManager.isDirty()).toBe(true); + dmManager.sealAndPublish.mockClear(); + await syncManager.pushSync(); + expect(dmManager.sealAndPublish).toHaveBeenCalled(); + }); + + it('pulls a push from another device met while confirming', async () => { + await syncManager.pushSync(); + const pullSpy = vi.spyOn(syncManager, 'pullSync').mockResolvedValue(null); + streamrController.fetchPartitionHistory.mockResolvedValue([other(4000), own(5000)]); + await vi.advanceTimersByTimeAsync(5000); + + expect(pullSpy).toHaveBeenCalled(); + }); + + it('pulls on a check only when another device pushed something newer', async () => { + streamrController.fetchPartitionHistory.mockResolvedValue([other(7000)]); + await syncManager.pullSync(); + const pullSpy = vi.spyOn(syncManager, 'pullSync').mockResolvedValue(null); + + expect(await syncManager.checkForNewSnapshot()).toBe(false); + streamrController.fetchPartitionHistory.mockResolvedValue([other(8000)]); + expect(await syncManager.checkForNewSnapshot()).toBe(true); + expect(streamrController.fetchPartitionHistory).toHaveBeenLastCalledWith('0xabc/Pombo-DM-1', 1, 1); + expect(pullSpy).toHaveBeenCalledTimes(1); + }); + + it('does not pull its own push back', async () => { + await syncManager.pushSync(); + const pullSpy = vi.spyOn(syncManager, 'pullSync'); + streamrController.fetchPartitionHistory.mockResolvedValue([own(5000)]); + + expect(await syncManager.checkForNewSnapshot()).toBe(false); + expect(pullSpy).not.toHaveBeenCalled(); + }); + + it('never checks in manual-only mode', async () => { + syncManager.setSyncMode(SYNC_MODES.MANUAL_ONLY); + streamrController.fetchPartitionHistory.mockResolvedValue([other(7000)]); + + expect(await syncManager.checkForNewSnapshot()).toBe(false); + expect(streamrController.fetchPartitionHistory).not.toHaveBeenCalled(); + }); + + it('checks once a minute while watched, and stops when told', async () => { + const checkSpy = vi.spyOn(syncManager, 'checkForNewSnapshot').mockResolvedValue(false); + syncManager.startSnapshotWatch(); + await vi.advanceTimersByTimeAsync(120000); + expect(checkSpy).toHaveBeenCalledTimes(2); + + syncManager.stopSnapshotWatch(); + await vi.advanceTimersByTimeAsync(60000); + expect(checkSpy).toHaveBeenCalledTimes(2); + }); + + it('leaves the blob partition a while after pushing blobs', async () => { + secureStorage.getUnsyncedImages.mockReturnValueOnce((async function* () { + yield { imageId: 'img-1', streamId: 's', encryptedData: 'x', iv: 'y' }; + })()); + await syncManager.pushImageBlobs(); + + expect(streamrController.leaveStreamPart).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(30000); + expect(streamrController.leaveStreamPart).toHaveBeenCalledWith('0xabc/Pombo-DM-1', 2); + }); + }); });