@@ -125,6 +125,10 @@ let snapshot: QueueSnapshot = { owner, items, synced };
125125let running : Promise < void > | null = null ;
126126// While Clear all asks the server, nothing replays: a queued note must not land in a Library being cleared.
127127let paused = 0 ;
128+ // Whether this account's stored queue has been read. Until it has, nothing is written over it.
129+ let hydrated = false ;
130+ // Writes that wait for the connection, told "queued" only once the queue holding them is saved.
131+ let awaitingSave : { key : string ; waiter : Waiter } [ ] = [ ] ;
128132let again = false ;
129133let keys = 0 ;
130134
@@ -163,12 +167,52 @@ function persist() {
163167 const gen = generation ;
164168 const saved = items . map ( ( item ) => ( { ...item } ) ) ;
165169 const savedBefore = { ...before } ;
170+ const confirming = awaitingSave ;
171+ awaitingSave = [ ] ;
166172 if ( ! userId ) return ;
167173 void storage ( async ( ) => {
168174 if ( gen !== generation ) return ;
169- if ( saved . length === 0 ) await AsyncStorage . removeItem ( storageKey ( userId ) ) ;
170- else await AsyncStorage . setItem ( storageKey ( userId ) , JSON . stringify ( { userId, items : saved , before : savedBefore } ) ) ;
171- } ) . catch ( warn ( 'Could not keep the offline changes' ) ) ;
175+ // A stored queue that could not be read is never written over: it may hold unsynced work.
176+ if ( ! hydrated ) throw new Error ( 'The offline changes on this phone could not be read.' ) ;
177+ const write = ( ) =>
178+ saved . length === 0
179+ ? AsyncStorage . removeItem ( storageKey ( userId ) )
180+ : AsyncStorage . setItem ( storageKey ( userId ) , JSON . stringify ( { userId, items : saved , before : savedBefore } ) ) ;
181+ // One retry for a passing storage error before anyone hears the change was not kept.
182+ await write ( ) . catch ( async ( ) => {
183+ await new Promise ( ( resolve ) => setTimeout ( resolve , 200 ) ) ;
184+ await write ( ) ;
185+ } ) ;
186+ } ) . then (
187+ ( ) => confirming . forEach ( ( { waiter } ) => waiter . resolve ( 'queued' ) ) ,
188+ ( cause : unknown ) => {
189+ warn ( 'Could not keep the offline changes' ) ( cause ) ;
190+ if ( gen === generation && confirming . length > 0 ) void unkeep ( userId , confirming , cause ) ;
191+ } ,
192+ ) ;
193+ }
194+
195+ /**
196+ * Writes that were to wait for the connection but could not be kept on the phone: they leave the
197+ * queue, the copies go back to what they showed, and whoever made them hears it, so a draft
198+ * stays open instead of being lost on the next launch.
199+ */
200+ async function unkeep ( userId : string , failed : { key : string ; waiter : Waiter } [ ] , cause : unknown ) {
201+ const keys = new Set ( failed . map ( ( one ) => one . key ) ) ;
202+ const gone = items . filter ( ( item ) => keys . has ( item . key ) && item . key !== inFlight ) ;
203+ for ( const item of gone ) {
204+ if ( item . kind === 'add' ) {
205+ dropWrites ( item . tempId , item . key ) ;
206+ updateCopy < Source [ ] > ( userId , 'sources' , item . input . debateId , ( list ) => list ?. filter ( ( source ) => source . id !== item . tempId ) ) ;
207+ } else {
208+ items = items . filter ( ( other ) => other . key !== item . key ) ;
209+ }
210+ }
211+ await copiesWritten ( ) ;
212+ for ( const item of gone ) if ( item . kind !== 'add' ) settleBefore ( userId , item , true ) ;
213+ const error = new Error ( 'This change could not be kept on this phone. Try again.' , { cause } ) ;
214+ failed . forEach ( ( { waiter } ) => waiter . reject ( error ) ) ;
215+ emit ( ) ;
172216}
173217
174218function parse ( raw : string | null , userId : string ) : { items : QueuedWrite [ ] ; before : Record < string , Restorable > } {
@@ -186,6 +230,32 @@ function parse(raw: string | null, userId: string): { items: QueuedWrite[]; befo
186230 }
187231}
188232
233+ /**
234+ * Reads the account's stored queue, once more after a passing failure. Writes from an earlier
235+ * launch go ahead of any made since; until a read succeeds nothing is written over the store.
236+ */
237+ async function hydrate ( userId : string , gen : number ) {
238+ let raw : string | null ;
239+ try {
240+ raw = await AsyncStorage . getItem ( storageKey ( userId ) ) . catch ( async ( ) => {
241+ await new Promise ( ( resolve ) => setTimeout ( resolve , 200 ) ) ;
242+ return AsyncStorage . getItem ( storageKey ( userId ) ) ;
243+ } ) ;
244+ } catch ( cause ) {
245+ warn ( 'Could not read the offline changes' ) ( cause ) ;
246+ return ;
247+ }
248+ if ( gen !== generation ) return ;
249+ const loaded = parse ( raw , userId ) ;
250+ hydrated = true ;
251+ if ( loaded . items . length === 0 ) return ;
252+ // Nobody waits for writes from an earlier launch: they show as waiting until they sync.
253+ const known = new Set ( items . map ( ( item ) => item . key ) ) ;
254+ items = [ ...loaded . items . filter ( ( item ) => ! known . has ( item . key ) ) . map ( ( item ) => ( { ...item , waiting : true } ) ) , ...items ] ;
255+ before = { ...loaded . before , ...before } ;
256+ emit ( ) ;
257+ }
258+
189259/**
190260 * Called with the signed-in account once auth has settled. Another account's
191261 * queue, or one left with no account, is removed; the same account after a
@@ -204,16 +274,13 @@ export function setQueueOwner(userId: string | null) {
204274 synced = new Map ( ) ;
205275 inFlight = null ;
206276 emit ( ) ;
277+ hydrated = false ;
278+ awaitingSave = [ ] ;
207279 ready = storage ( async ( ) => {
208280 const stale = ( await AsyncStorage . getAllKeys ( ) ) . filter ( ( key ) => key . startsWith ( `${ PREFIX } :` ) && ( userId === null || key !== storageKey ( userId ) ) ) ;
209281 if ( stale . length > 0 ) await AsyncStorage . multiRemove ( stale ) ;
210282 if ( ! userId ) return ;
211- const loaded = parse ( await AsyncStorage . getItem ( storageKey ( userId ) ) , userId ) ;
212- if ( gen !== generation || loaded . items . length === 0 ) return ;
213- // Nobody waits for writes from an earlier launch: they show as waiting until they sync.
214- items = loaded . items . map ( ( item ) => ( { ...item , waiting : true } ) ) ;
215- before = loaded . before ;
216- emit ( ) ;
283+ await hydrate ( userId , gen ) ;
217284 } ) . catch ( warn ( 'Could not read the offline changes' ) ) ;
218285}
219286
@@ -223,13 +290,20 @@ async function adopt(userId: string) {
223290 else if ( owner !== userId ) throw new Error ( 'This change belongs to another account.' ) ;
224291 await ready ;
225292 if ( owner !== userId ) throw new Error ( 'This change belongs to another account.' ) ;
293+ // The stored queue could not be read at launch: try again before adding to it.
294+ if ( ! hydrated ) {
295+ const gen = generation ;
296+ ready = storage ( ( ) => hydrate ( userId , gen ) ) . catch ( warn ( 'Could not read the offline changes' ) ) ;
297+ await ready ;
298+ }
226299}
227300
228301const isWaitingForNetwork = ( ) => isKnownOffline ( ) || items . some ( ( item ) => item . waiting ) ;
229302
230303function attach ( key : string , waiter : Waiter , waiting : boolean ) {
231304 if ( waiting ) {
232- waiter . resolve ( 'queued' ) ;
305+ // Told "queued" once the queue holding it is saved (persist), never before.
306+ awaitingSave . push ( { key, waiter } ) ;
233307 return ;
234308 }
235309 waiters . set ( key , [ ...( waiters . get ( key ) ?? [ ] ) , waiter ] ) ;
@@ -247,7 +321,7 @@ function parkAll() {
247321 items = items . map ( ( item ) => ( item . waiting ? item : { ...item , waiting : true } ) ) ;
248322 const parked = waiters ;
249323 waiters = new Map ( ) ;
250- parked . forEach ( ( list ) => list . forEach ( ( waiter ) => waiter . resolve ( 'queued' ) ) ) ;
324+ parked . forEach ( ( list , key ) => list . forEach ( ( waiter ) => awaitingSave . push ( { key , waiter } ) ) ) ;
251325 persist ( ) ;
252326 emit ( ) ;
253327}
0 commit comments