fix: prevent offline queue replay data loss
This commit is contained in:
@@ -221,6 +221,7 @@ export const AppProvider = ({ children }) => {
|
||||
if (getStoredUser()?.email !== owner) throw new Error('Oturum değişti; kuyruk bekletildi.');
|
||||
savedTenderStates.current = { ...savedTenderStates.current, [item.key]: record };
|
||||
setTenders(prev => applyTenderStates(prev, savedTenderStates.current));
|
||||
return record;
|
||||
});
|
||||
setOfflineQueueSize(offlineMutationCount(owner));
|
||||
if (result.error) setSaveError(`Çevrimdışı kayıt gönderilemedi: ${result.error.message}`);
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
const PREFIX = 'ihaleist_offline_mutations:';
|
||||
const LIMIT = 100;
|
||||
const activeReplays = new Map();
|
||||
|
||||
const ownerKey = owner => `${PREFIX}${encodeURIComponent(String(owner || '').trim().toLowerCase())}`;
|
||||
|
||||
@@ -14,7 +15,7 @@ function readStorage(owner) {
|
||||
function writeStorage(owner, rows) {
|
||||
if (typeof localStorage === 'undefined' || !String(owner || '').trim()) return false;
|
||||
try {
|
||||
localStorage.setItem(ownerKey(owner), JSON.stringify(rows.slice(-LIMIT)));
|
||||
localStorage.setItem(ownerKey(owner), JSON.stringify(rows));
|
||||
return true;
|
||||
} catch { return false; }
|
||||
}
|
||||
@@ -39,6 +40,9 @@ export function enqueueTenderMutation(owner, { key, patch, snapshot, version })
|
||||
const rows = readStorage(owner);
|
||||
const index = rows.findIndex(item => item.kind === 'tender-state' && item.key === key);
|
||||
const previous = index >= 0 ? rows[index] : null;
|
||||
if (!previous && rows.length >= LIMIT) {
|
||||
throw new Error('Çevrimdışı kayıt kuyruğu dolu. Bağlantı kurulduktan sonra yeniden deneyin.');
|
||||
}
|
||||
const next = {
|
||||
id: previous?.id || `offline-${crypto.randomUUID()}`,
|
||||
kind: 'tender-state',
|
||||
@@ -49,6 +53,7 @@ export function enqueueTenderMutation(owner, { key, patch, snapshot, version })
|
||||
createdAt: previous?.createdAt || new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
attempts: previous?.attempts || 0,
|
||||
revision: (Number.isInteger(previous?.revision) ? previous.revision : 0) + 1,
|
||||
};
|
||||
if (index >= 0) rows[index] = next; else rows.push(next);
|
||||
if (!writeStorage(owner, rows)) throw new Error('Çevrimdışı kayıt kuyruğu yazılamadı.');
|
||||
@@ -67,27 +72,56 @@ export function markOfflineMutationAttempt(owner, id) {
|
||||
return next.find(item => item.id === id) || null;
|
||||
}
|
||||
|
||||
/** Replay in order; a failed item remains queued and stops the batch. */
|
||||
export async function replayOfflineMutations(owner, send) {
|
||||
function acknowledgeOfflineMutation(owner, item, record) {
|
||||
const rows = readStorage(owner);
|
||||
const index = rows.findIndex(row => row.id === item.id);
|
||||
if (index < 0) return true;
|
||||
const current = rows[index];
|
||||
if ((current.revision || 0) === (item.revision || 0)) {
|
||||
rows.splice(index, 1);
|
||||
} else {
|
||||
const nextVersion = Number(record?.version);
|
||||
rows[index] = {
|
||||
...current,
|
||||
version: Number.isInteger(nextVersion) && nextVersion >= 0 ? nextVersion : current.version,
|
||||
};
|
||||
}
|
||||
return writeStorage(owner, rows);
|
||||
}
|
||||
|
||||
async function runReplay(owner, send) {
|
||||
const rows = listOfflineMutations(owner);
|
||||
const result = { sent: 0, pending: rows.length, error: null };
|
||||
for (const item of rows) {
|
||||
markOfflineMutationAttempt(owner, item.id);
|
||||
try {
|
||||
await send(item);
|
||||
removeOfflineMutation(owner, item.id);
|
||||
const record = await send(item);
|
||||
if (!acknowledgeOfflineMutation(owner, item, record)) {
|
||||
throw new Error('Çevrimdışı kayıt kuyruğu güncellenemedi.');
|
||||
}
|
||||
result.sent += 1;
|
||||
result.pending -= 1;
|
||||
} catch (error) {
|
||||
result.error = error;
|
||||
break;
|
||||
}
|
||||
}
|
||||
result.pending = offlineMutationCount(owner);
|
||||
return result;
|
||||
}
|
||||
|
||||
/** Replay in order; concurrent reconnect signals share one replay operation. */
|
||||
export function replayOfflineMutations(owner, send) {
|
||||
const replayKey = ownerKey(owner);
|
||||
const active = activeReplays.get(replayKey);
|
||||
if (active) return active;
|
||||
const replay = runReplay(owner, send).finally(() => {
|
||||
if (activeReplays.get(replayKey) === replay) activeReplays.delete(replayKey);
|
||||
});
|
||||
activeReplays.set(replayKey, replay);
|
||||
return replay;
|
||||
}
|
||||
|
||||
export function clearOfflineMutations(owner) {
|
||||
if (typeof localStorage === 'undefined') return false;
|
||||
try { localStorage.removeItem(ownerKey(owner)); return true; } catch { return false; }
|
||||
}
|
||||
|
||||
|
||||
@@ -43,3 +43,49 @@ test('replay removes acknowledged rows and retains the first failed row', async
|
||||
clearOfflineMutations('replay@example.test');
|
||||
});
|
||||
|
||||
test('concurrent reconnect signals send each queued mutation once', async () => {
|
||||
clearOfflineMutations('concurrent@example.test');
|
||||
enqueueTenderMutation('concurrent@example.test', { key: 'a', patch: { takipte: true }, snapshot: {}, version: 0 });
|
||||
let sends = 0;
|
||||
let release;
|
||||
const gate = new Promise(resolve => { release = resolve; });
|
||||
const send = async () => { sends += 1; await gate; return { version: 1 }; };
|
||||
const first = replayOfflineMutations('concurrent@example.test', send);
|
||||
const second = replayOfflineMutations('concurrent@example.test', send);
|
||||
release();
|
||||
const [a, b] = await Promise.all([first, second]);
|
||||
assert.equal(sends, 1);
|
||||
assert.deepEqual(a, b);
|
||||
assert.equal(offlineMutationCount('concurrent@example.test'), 0);
|
||||
});
|
||||
|
||||
test('a mutation added during replay is retained with the acknowledged server version', async () => {
|
||||
clearOfflineMutations('during-replay@example.test');
|
||||
enqueueTenderMutation('during-replay@example.test', { key: 'a', patch: { takipte: true }, snapshot: {}, version: 3 });
|
||||
const first = await replayOfflineMutations('during-replay@example.test', async item => {
|
||||
enqueueTenderMutation('during-replay@example.test', { key: 'a', patch: { kanbanStatus: 'teklif_hazirlaniyor' }, snapshot: {}, version: item.version });
|
||||
return { version: 4 };
|
||||
});
|
||||
assert.equal(first.pending, 1);
|
||||
const retained = listOfflineMutations('during-replay@example.test')[0];
|
||||
assert.equal(retained.version, 4);
|
||||
assert.deepEqual(retained.patch, { takipte: true, kanbanStatus: 'teklif_hazirlaniyor' });
|
||||
const versions = [];
|
||||
await replayOfflineMutations('during-replay@example.test', async item => { versions.push(item.version); return { version: 5 }; });
|
||||
assert.deepEqual(versions, [4]);
|
||||
assert.equal(offlineMutationCount('during-replay@example.test'), 0);
|
||||
});
|
||||
|
||||
test('a full queue rejects a new tender instead of silently discarding an older change', () => {
|
||||
clearOfflineMutations('full@example.test');
|
||||
for (let index = 0; index < 100; index += 1) {
|
||||
enqueueTenderMutation('full@example.test', { key: String(index), patch: { takipte: true }, snapshot: {}, version: 0 });
|
||||
}
|
||||
assert.throws(
|
||||
() => enqueueTenderMutation('full@example.test', { key: 'overflow', patch: { takipte: true }, snapshot: {}, version: 0 }),
|
||||
/kuyruğu dolu/,
|
||||
);
|
||||
assert.equal(offlineMutationCount('full@example.test'), 100);
|
||||
assert.equal(listOfflineMutations('full@example.test')[0].key, '0');
|
||||
clearOfflineMutations('full@example.test');
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user