mirror of
https://github.com/johannesjo/super-productivity.git
synced 2026-07-17 16:37:43 +00:00
fix(sync): serialize archive mutations and persist before dispatch
Remote archive side effects must be idempotent and immune to concurrent read-modify-write races, or a retried operation can duplicate or drop archive rows. - TaskArchiveService serializes every archive mutation behind a dedicated sp_task_archive web lock (separate from OPERATION_LOG, which remote archive handlers already hold non-reentrantly). - ArchiveService.moveTasksToArchiveAndFlushArchiveIfDue writes tasks with setMany over a deduplicated sorted id set, so a retry over a partially written archive converges instead of double-adding. - TaskService._moveToArchive coalesces concurrent archive calls for the same ids and persists archive data before dispatching moveToArchive, so a full-state snapshot cannot acknowledge the op while its archive write is still in flight.
This commit is contained in:
Johannes Millan
parent
ffca0f1397
commit
982f2916ca
7 changed files with 464 additions and 49 deletions
|
|
@ -83,6 +83,86 @@ describe('ArchiveService', () => {
|
|||
expect(savedData.task.ids).toContain('task-1');
|
||||
});
|
||||
|
||||
it('should serialize concurrent task archive mutations without dropping either task', async () => {
|
||||
let archiveYoung = createEmptyArchive(Date.now());
|
||||
let releaseFirstSave: (() => void) | undefined;
|
||||
let firstSaveStarted: (() => void) | undefined;
|
||||
const firstSaveStartedPromise = new Promise<void>((resolve) => {
|
||||
firstSaveStarted = resolve;
|
||||
});
|
||||
const firstSaveGate = new Promise<void>((resolve) => {
|
||||
releaseFirstSave = resolve;
|
||||
});
|
||||
|
||||
mockArchiveDbAdapter.loadArchiveYoung.and.callFake(async () => archiveYoung);
|
||||
mockArchiveDbAdapter.loadArchiveOld.and.callFake(async () =>
|
||||
createEmptyArchive(Date.now()),
|
||||
);
|
||||
mockArchiveDbAdapter.saveArchiveYoung.and.callFake(async (nextArchive) => {
|
||||
if (mockArchiveDbAdapter.saveArchiveYoung.calls.count() === 1) {
|
||||
firstSaveStarted?.();
|
||||
await firstSaveGate;
|
||||
}
|
||||
archiveYoung = nextArchive;
|
||||
});
|
||||
|
||||
const first = service.moveTasksToArchiveAndFlushArchiveIfDue([
|
||||
createMockTask('task-a'),
|
||||
]);
|
||||
await firstSaveStartedPromise;
|
||||
const second = service.moveTasksToArchiveAndFlushArchiveIfDue([
|
||||
createMockTask('task-b'),
|
||||
]);
|
||||
|
||||
// The second read must wait for the first write to commit.
|
||||
await Promise.resolve();
|
||||
expect(mockArchiveDbAdapter.loadArchiveYoung).toHaveBeenCalledTimes(1);
|
||||
|
||||
releaseFirstSave?.();
|
||||
await Promise.all([first, second]);
|
||||
|
||||
expect(archiveYoung.task.ids).toEqual(['task-a', 'task-b']);
|
||||
expect(Object.keys(archiveYoung.task.entities).sort()).toEqual([
|
||||
'task-a',
|
||||
'task-b',
|
||||
]);
|
||||
});
|
||||
|
||||
it('should replace a stale archived task on retry and keep ids deduped and sorted', async () => {
|
||||
const staleTask = createMockTask('task-b', {
|
||||
title: 'Stale archived title',
|
||||
timeSpent: 1,
|
||||
});
|
||||
const otherTask = createMockTask('task-a');
|
||||
mockArchiveDbAdapter.loadArchiveYoung.and.returnValue(
|
||||
Promise.resolve({
|
||||
...createEmptyArchive(),
|
||||
task: {
|
||||
ids: ['task-b', 'task-a', 'task-b'],
|
||||
entities: {
|
||||
[otherTask.id]: otherTask,
|
||||
[staleTask.id]: staleTask,
|
||||
},
|
||||
},
|
||||
}),
|
||||
);
|
||||
mockArchiveDbAdapter.loadArchiveOld.and.returnValue(
|
||||
Promise.resolve(createEmptyArchive(Date.now() - 1000)),
|
||||
);
|
||||
const retriedTask = createMockTask('task-b', {
|
||||
title: 'Newest active title',
|
||||
timeSpent: 42,
|
||||
});
|
||||
|
||||
await service.moveTasksToArchiveAndFlushArchiveIfDue([retriedTask]);
|
||||
|
||||
const savedTaskState =
|
||||
mockArchiveDbAdapter.saveArchiveYoung.calls.first().args[0].task;
|
||||
expect(savedTaskState.ids).toEqual(['task-a', 'task-b']);
|
||||
expect(savedTaskState.entities['task-b']!.title).toBe('Newest active title');
|
||||
expect(savedTaskState.entities['task-b']!.timeSpent).toBe(42);
|
||||
});
|
||||
|
||||
it('should dispatch updateWholeState for time tracking', async () => {
|
||||
const tasks = [createMockTask('task-1')];
|
||||
|
||||
|
|
@ -344,6 +424,38 @@ describe('ArchiveService', () => {
|
|||
});
|
||||
|
||||
describe('writeTasksToArchiveForRemoteSync', () => {
|
||||
it('should replace a stale archived task on remote retry and keep ids deduped and sorted', async () => {
|
||||
const staleTask = createMockTask('task-b', {
|
||||
title: 'Stale remote title',
|
||||
timeSpent: 1,
|
||||
});
|
||||
const otherTask = createMockTask('task-a');
|
||||
mockArchiveDbAdapter.loadArchiveYoung.and.returnValue(
|
||||
Promise.resolve({
|
||||
...createEmptyArchive(),
|
||||
task: {
|
||||
ids: ['task-b', 'task-a', 'task-b'],
|
||||
entities: {
|
||||
[otherTask.id]: otherTask,
|
||||
[staleTask.id]: staleTask,
|
||||
},
|
||||
},
|
||||
}),
|
||||
);
|
||||
const retriedTask = createMockTask('task-b', {
|
||||
title: 'Newest remote title',
|
||||
timeSpent: 84,
|
||||
});
|
||||
|
||||
await service.writeTasksToArchiveForRemoteSync([retriedTask]);
|
||||
|
||||
const savedTaskState =
|
||||
mockArchiveDbAdapter.saveArchiveYoung.calls.mostRecent().args[0].task;
|
||||
expect(savedTaskState.ids).toEqual(['task-a', 'task-b']);
|
||||
expect(savedTaskState.entities['task-b']!.title).toBe('Newest remote title');
|
||||
expect(savedTaskState.entities['task-b']!.timeSpent).toBe(84);
|
||||
});
|
||||
|
||||
it('should ignore malformed tasks without valid ids', async () => {
|
||||
const invalidTask = {
|
||||
title: 'Invalid task',
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
import { inject, Injectable } from '@angular/core';
|
||||
import { Task, TaskWithSubTasks } from '../tasks/task.model';
|
||||
import { Task, TaskArchive, TaskWithSubTasks } from '../tasks/task.model';
|
||||
import { flattenTasks } from '../tasks/store/task.selectors';
|
||||
import { createEmptyEntity } from '../../util/create-empty-entity';
|
||||
import { taskAdapter } from '../tasks/store/task.adapter';
|
||||
|
|
@ -19,6 +19,8 @@ import { first } from 'rxjs/operators';
|
|||
import { firstValueFrom } from 'rxjs';
|
||||
import { selectTimeTrackingState } from '../time-tracking/store/time-tracking.selectors';
|
||||
import { isValidEntityId } from '../../op-log/validation/is-valid-entity-id';
|
||||
import { LockService } from '../../op-log/sync/lock.service';
|
||||
import { LOCK_NAMES } from '../../op-log/core/operation-log.const';
|
||||
|
||||
/**
|
||||
* Maps tasks to archive format by:
|
||||
|
|
@ -60,6 +62,23 @@ const mapTasksToArchiveFormat = (
|
|||
});
|
||||
};
|
||||
|
||||
const replaceTasksInArchive = (
|
||||
archiveTasks: Task[],
|
||||
taskArchiveState: TaskArchive,
|
||||
): TaskArchive => {
|
||||
const updatedTaskArchive = taskAdapter.setMany(archiveTasks, taskArchiveState);
|
||||
const uniqueStringIds = new Set(
|
||||
updatedTaskArchive.ids.filter((id): id is string => typeof id === 'string'),
|
||||
);
|
||||
|
||||
return {
|
||||
...updatedTaskArchive,
|
||||
// Keep retry output deterministic even when an older archive already
|
||||
// contains duplicate or unsorted IDs.
|
||||
ids: [...uniqueStringIds].sort(),
|
||||
};
|
||||
};
|
||||
|
||||
export const sanitizeTasksForArchiving = (
|
||||
tasksIn: TaskWithSubTasks[],
|
||||
logPrefix: string,
|
||||
|
|
@ -153,10 +172,29 @@ const DEFAULT_ARCHIVE: ArchiveModel = {
|
|||
export class ArchiveService {
|
||||
private readonly _archiveDbAdapter = inject(ArchiveDbAdapter);
|
||||
private readonly _store = inject(Store);
|
||||
private readonly _lockService = inject(LockService);
|
||||
|
||||
/**
|
||||
* Serializes task archive read-modify-write cycles. Separate task actions can
|
||||
* arrive before either IndexedDB write finishes; without one shared queue,
|
||||
* both calls can read the same archive and the last save silently drops the
|
||||
* other task. A failed mutation does not poison later retries.
|
||||
*/
|
||||
private _runTaskArchiveMutation(mutation: () => Promise<void>): Promise<void> {
|
||||
return this._lockService.request(LOCK_NAMES.TASK_ARCHIVE, mutation);
|
||||
}
|
||||
|
||||
// NOTE: we choose this method as trigger to check for flushing to archive, since
|
||||
// it is usually triggered every work-day once
|
||||
async moveTasksToArchiveAndFlushArchiveIfDue(tasks: TaskWithSubTasks[]): Promise<void> {
|
||||
moveTasksToArchiveAndFlushArchiveIfDue(tasks: TaskWithSubTasks[]): Promise<void> {
|
||||
return this._runTaskArchiveMutation(() =>
|
||||
this._moveTasksToArchiveAndFlushArchiveIfDue(tasks),
|
||||
);
|
||||
}
|
||||
|
||||
private async _moveTasksToArchiveAndFlushArchiveIfDue(
|
||||
tasks: TaskWithSubTasks[],
|
||||
): Promise<void> {
|
||||
const now = Date.now();
|
||||
const sanitizedTasks = sanitizeTasksForArchiving(tasks, 'moveToArchive');
|
||||
const flatTasks = flattenTasks(sanitizedTasks);
|
||||
|
|
@ -178,12 +216,7 @@ export class ArchiveService {
|
|||
const taskArchiveState = archiveYoung.task || createEmptyEntity();
|
||||
|
||||
const archiveTasks = mapTasksToArchiveFormat(flatTasks, now, 'moveToArchive');
|
||||
const newTaskArchiveUnsorted = taskAdapter.addMany(archiveTasks, taskArchiveState);
|
||||
// Sort ids for deterministic ordering across clients (UUIDv7 is lexicographically sortable)
|
||||
const newTaskArchive = {
|
||||
...newTaskArchiveUnsorted,
|
||||
ids: [...newTaskArchiveUnsorted.ids].sort(),
|
||||
};
|
||||
const newTaskArchive = replaceTasksInArchive(archiveTasks, taskArchiveState);
|
||||
|
||||
// ------------------------------------------------
|
||||
// Result A:
|
||||
|
|
@ -317,7 +350,15 @@ export class ArchiveService {
|
|||
* only on the originating client and synced via the flushYoungToOld action.
|
||||
* This ensures flushes happen exactly once, not on every receiving client.
|
||||
*/
|
||||
async writeTasksToArchiveForRemoteSync(tasks: TaskWithSubTasks[]): Promise<void> {
|
||||
writeTasksToArchiveForRemoteSync(tasks: TaskWithSubTasks[]): Promise<void> {
|
||||
return this._runTaskArchiveMutation(() =>
|
||||
this._writeTasksToArchiveForRemoteSync(tasks),
|
||||
);
|
||||
}
|
||||
|
||||
private async _writeTasksToArchiveForRemoteSync(
|
||||
tasks: TaskWithSubTasks[],
|
||||
): Promise<void> {
|
||||
const now = Date.now();
|
||||
const sanitizedTasks = sanitizeTasksForArchiving(tasks, 'Remote sync');
|
||||
const flatTasks = flattenTasks(sanitizedTasks);
|
||||
|
|
@ -339,12 +380,7 @@ export class ArchiveService {
|
|||
const taskArchiveState = archiveYoung.task || createEmptyEntity();
|
||||
|
||||
const archiveTasks = mapTasksToArchiveFormat(flatTasks, now, 'Remote sync');
|
||||
const newTaskArchiveUnsorted = taskAdapter.addMany(archiveTasks, taskArchiveState);
|
||||
// Sort ids for deterministic ordering across clients (UUIDv7 is lexicographically sortable)
|
||||
const newTaskArchive = {
|
||||
...newTaskArchiveUnsorted,
|
||||
ids: [...newTaskArchiveUnsorted.ids].sort(),
|
||||
};
|
||||
const newTaskArchive = replaceTasksInArchive(archiveTasks, taskArchiveState);
|
||||
|
||||
// Also move historical time tracking data to archiveYoung
|
||||
// This ensures the remote client's archive matches the originating client
|
||||
|
|
|
|||
|
|
@ -6,11 +6,39 @@ import { Task, TaskArchive, TaskState } from '../tasks/task.model';
|
|||
import { ArchiveModel } from './archive.model';
|
||||
import { Update } from '@ngrx/entity';
|
||||
import { RoundTimeOption } from '../project/project.model';
|
||||
import { ArchiveService } from './archive.service';
|
||||
import { of } from 'rxjs';
|
||||
import { LockService } from '../../op-log/sync/lock.service';
|
||||
|
||||
class TestLockService {
|
||||
private readonly _tails = new Map<string, Promise<void>>();
|
||||
|
||||
async request<T>(lockName: string, callback: () => Promise<T>): Promise<T> {
|
||||
const previous = this._tails.get(lockName) ?? Promise.resolve();
|
||||
let release: (() => void) | undefined;
|
||||
const current = new Promise<void>((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
const tail = previous.then(() => current);
|
||||
this._tails.set(lockName, tail);
|
||||
|
||||
await previous;
|
||||
try {
|
||||
return await callback();
|
||||
} finally {
|
||||
release?.();
|
||||
if (this._tails.get(lockName) === tail) {
|
||||
this._tails.delete(lockName);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
import { TaskSharedActions } from '../../root-store/meta/task-shared.actions';
|
||||
|
||||
describe('TaskArchiveService', () => {
|
||||
let service: TaskArchiveService;
|
||||
let archiveService: ArchiveService;
|
||||
let storeMock: jasmine.SpyObj<Store>;
|
||||
let archiveDbAdapterMock: jasmine.SpyObj<ArchiveDbAdapter>;
|
||||
|
||||
|
|
@ -53,7 +81,8 @@ describe('TaskArchiveService', () => {
|
|||
});
|
||||
|
||||
beforeEach(() => {
|
||||
storeMock = jasmine.createSpyObj<Store>('Store', ['dispatch']);
|
||||
storeMock = jasmine.createSpyObj<Store>('Store', ['dispatch', 'select']);
|
||||
storeMock.select.and.returnValue(of({ project: {}, tag: {} }));
|
||||
|
||||
archiveDbAdapterMock = jasmine.createSpyObj<ArchiveDbAdapter>('ArchiveDbAdapter', [
|
||||
'loadArchiveYoung',
|
||||
|
|
@ -75,10 +104,58 @@ describe('TaskArchiveService', () => {
|
|||
TaskArchiveService,
|
||||
{ provide: ArchiveDbAdapter, useValue: archiveDbAdapterMock },
|
||||
{ provide: Store, useValue: storeMock },
|
||||
{ provide: LockService, useClass: TestLockService },
|
||||
],
|
||||
});
|
||||
|
||||
service = TestBed.inject(TaskArchiveService);
|
||||
archiveService = TestBed.inject(ArchiveService);
|
||||
});
|
||||
|
||||
describe('archive mutation locking', () => {
|
||||
it('should serialize a task update with an ArchiveService mutation', async () => {
|
||||
const existingTask = createMockTask('existing', { title: 'Original' });
|
||||
const archivedTask = {
|
||||
...createMockTask('new-task', {
|
||||
isDone: true,
|
||||
doneOn: Date.now(),
|
||||
}),
|
||||
subTasks: [],
|
||||
};
|
||||
let archiveYoung = createMockArchiveModel([existingTask]);
|
||||
let releaseFirstSave: (() => void) | undefined;
|
||||
let firstSaveStarted: (() => void) | undefined;
|
||||
const firstSaveStartedPromise = new Promise<void>((resolve) => {
|
||||
firstSaveStarted = resolve;
|
||||
});
|
||||
const firstSaveGate = new Promise<void>((resolve) => {
|
||||
releaseFirstSave = resolve;
|
||||
});
|
||||
|
||||
archiveDbAdapterMock.loadArchiveYoung.and.callFake(async () => archiveYoung);
|
||||
archiveDbAdapterMock.saveArchiveYoung.and.callFake(async (nextArchive) => {
|
||||
if (archiveDbAdapterMock.saveArchiveYoung.calls.count() === 1) {
|
||||
firstSaveStarted?.();
|
||||
await firstSaveGate;
|
||||
}
|
||||
archiveYoung = nextArchive;
|
||||
});
|
||||
|
||||
const updatePromise = service.updateTask('existing', { title: 'Updated' });
|
||||
await firstSaveStartedPromise;
|
||||
const archivePromise = archiveService.moveTasksToArchiveAndFlushArchiveIfDue([
|
||||
archivedTask,
|
||||
]);
|
||||
|
||||
await new Promise<void>((resolve) => setTimeout(resolve, 0));
|
||||
expect(archiveDbAdapterMock.loadArchiveYoung).toHaveBeenCalledTimes(1);
|
||||
|
||||
releaseFirstSave?.();
|
||||
await Promise.all([updatePromise, archivePromise]);
|
||||
|
||||
expect(archiveYoung.task.entities['existing']!.title).toBe('Updated');
|
||||
expect(archiveYoung.task.entities['new-task']).toBeDefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe('loadYoung', () => {
|
||||
|
|
|
|||
|
|
@ -17,6 +17,8 @@ import { TAG_FEATURE_NAME, tagAdapter } from '../tag/store/tag.reducer';
|
|||
import { WORK_CONTEXT_FEATURE_NAME } from '../work-context/store/work-context.selectors';
|
||||
import { plannerFeatureKey } from '../planner/store/planner.reducer';
|
||||
import { TODAY_TAG } from '../tag/tag.const';
|
||||
import { LockService } from '../../op-log/sync/lock.service';
|
||||
import { LOCK_NAMES } from '../../op-log/core/operation-log.const';
|
||||
|
||||
// Normalize timeSpentOnDay at the data boundary so all consumers can trust the
|
||||
// invariant: timeSpentOnDay is always a valid object, never undefined. This mirrors
|
||||
|
|
@ -74,6 +76,7 @@ type TaskArchiveAction =
|
|||
})
|
||||
export class TaskArchiveService {
|
||||
private _injector = inject(Injector);
|
||||
private readonly _lockService = inject(LockService);
|
||||
private _archiveDbAdapter?: ArchiveDbAdapter;
|
||||
private get archiveDbAdapter(): ArchiveDbAdapter {
|
||||
if (!this._archiveDbAdapter) {
|
||||
|
|
@ -106,6 +109,15 @@ export class TaskArchiveService {
|
|||
|
||||
constructor() {}
|
||||
|
||||
private _runTaskArchiveMutation(
|
||||
mutation: () => Promise<void>,
|
||||
isIgnoreDBLock: boolean = false,
|
||||
): Promise<void> {
|
||||
return isIgnoreDBLock
|
||||
? mutation()
|
||||
: this._lockService.request(LOCK_NAMES.TASK_ARCHIVE, mutation);
|
||||
}
|
||||
|
||||
async loadYoung(): Promise<TaskArchive> {
|
||||
const archiveYoung =
|
||||
(await this.archiveDbAdapter.loadArchiveYoung()) || DEFAULT_ARCHIVE;
|
||||
|
|
@ -230,10 +242,17 @@ export class TaskArchiveService {
|
|||
return result;
|
||||
}
|
||||
|
||||
async deleteTasks(
|
||||
deleteTasks(
|
||||
taskIdsToDelete: string[],
|
||||
options?: { isIgnoreDBLock?: boolean },
|
||||
): Promise<void> {
|
||||
return this._runTaskArchiveMutation(
|
||||
() => this._deleteTasks(taskIdsToDelete),
|
||||
options?.isIgnoreDBLock,
|
||||
);
|
||||
}
|
||||
|
||||
private async _deleteTasks(taskIdsToDelete: string[]): Promise<void> {
|
||||
const archiveYoung =
|
||||
(await this.archiveDbAdapter.loadArchiveYoung()) || DEFAULT_ARCHIVE;
|
||||
const toDeleteInArchiveYoung = taskIdsToDelete.filter(
|
||||
|
|
@ -272,7 +291,18 @@ export class TaskArchiveService {
|
|||
}
|
||||
}
|
||||
|
||||
async updateTask(
|
||||
updateTask(
|
||||
id: string,
|
||||
changedFields: Partial<Task>,
|
||||
options?: { isSkipDispatch?: boolean; isIgnoreDBLock?: boolean },
|
||||
): Promise<void> {
|
||||
return this._runTaskArchiveMutation(
|
||||
() => this._updateTask(id, changedFields, options),
|
||||
options?.isIgnoreDBLock,
|
||||
);
|
||||
}
|
||||
|
||||
private async _updateTask(
|
||||
id: string,
|
||||
changedFields: Partial<Task>,
|
||||
options?: { isSkipDispatch?: boolean; isIgnoreDBLock?: boolean },
|
||||
|
|
@ -311,7 +341,17 @@ export class TaskArchiveService {
|
|||
throw new Error('Archive task to update not found');
|
||||
}
|
||||
|
||||
async updateTasks(
|
||||
updateTasks(
|
||||
updates: Update<Task>[],
|
||||
options?: { isSkipDispatch?: boolean; isIgnoreDBLock?: boolean },
|
||||
): Promise<void> {
|
||||
return this._runTaskArchiveMutation(
|
||||
() => this._updateTasks(updates, options),
|
||||
options?.isIgnoreDBLock,
|
||||
);
|
||||
}
|
||||
|
||||
private async _updateTasks(
|
||||
updates: Update<Task>[],
|
||||
options?: { isSkipDispatch?: boolean; isIgnoreDBLock?: boolean },
|
||||
): Promise<void> {
|
||||
|
|
@ -364,9 +404,18 @@ export class TaskArchiveService {
|
|||
}
|
||||
|
||||
// -----------------------------------------
|
||||
async removeAllArchiveTasksForProject(
|
||||
removeAllArchiveTasksForProject(
|
||||
projectIdToDelete: string,
|
||||
options?: { isIgnoreDBLock?: boolean },
|
||||
): Promise<void> {
|
||||
return this._runTaskArchiveMutation(
|
||||
() => this._removeAllArchiveTasksForProject(projectIdToDelete),
|
||||
options?.isIgnoreDBLock,
|
||||
);
|
||||
}
|
||||
|
||||
private async _removeAllArchiveTasksForProject(
|
||||
projectIdToDelete: string,
|
||||
): Promise<void> {
|
||||
const taskArchiveState: TaskArchive = await this.load();
|
||||
const archiveTaskIdsToDelete = !!taskArchiveState
|
||||
|
|
@ -378,13 +427,20 @@ export class TaskArchiveService {
|
|||
return t.projectId === projectIdToDelete;
|
||||
})
|
||||
: [];
|
||||
await this.deleteTasks(archiveTaskIdsToDelete, options);
|
||||
await this._deleteTasks(archiveTaskIdsToDelete);
|
||||
}
|
||||
|
||||
async removeTagsFromAllTasks(
|
||||
removeTagsFromAllTasks(
|
||||
tagIdsToRemove: string[],
|
||||
options?: { isIgnoreDBLock?: boolean },
|
||||
): Promise<void> {
|
||||
return this._runTaskArchiveMutation(
|
||||
() => this._removeTagsFromAllTasks(tagIdsToRemove),
|
||||
options?.isIgnoreDBLock,
|
||||
);
|
||||
}
|
||||
|
||||
private async _removeTagsFromAllTasks(tagIdsToRemove: string[]): Promise<void> {
|
||||
const taskArchiveState: TaskArchive = await this.load();
|
||||
await this._execActionBoth(
|
||||
TaskSharedActions.removeTagsForAllTasks({ tagIdsToRemove }),
|
||||
|
|
@ -406,16 +462,23 @@ export class TaskArchiveService {
|
|||
}
|
||||
});
|
||||
// TODO check to maybe update to today tag instead
|
||||
await this.deleteTasks(
|
||||
[...archiveMainTaskIdsToDelete, ...archiveSubTaskIdsToDelete],
|
||||
options,
|
||||
);
|
||||
await this._deleteTasks([
|
||||
...archiveMainTaskIdsToDelete,
|
||||
...archiveSubTaskIdsToDelete,
|
||||
]);
|
||||
}
|
||||
|
||||
async removeRepeatCfgFromArchiveTasks(
|
||||
removeRepeatCfgFromArchiveTasks(
|
||||
repeatConfigId: string,
|
||||
options?: { isIgnoreDBLock?: boolean },
|
||||
): Promise<void> {
|
||||
return this._runTaskArchiveMutation(
|
||||
() => this._removeRepeatCfgFromArchiveTasks(repeatConfigId),
|
||||
options?.isIgnoreDBLock,
|
||||
);
|
||||
}
|
||||
|
||||
private async _removeRepeatCfgFromArchiveTasks(repeatConfigId: string): Promise<void> {
|
||||
const taskArchive = await this.load();
|
||||
|
||||
const newState = { ...taskArchive };
|
||||
|
|
@ -435,16 +498,24 @@ export class TaskArchiveService {
|
|||
},
|
||||
};
|
||||
});
|
||||
await this.updateTasks(updates, {
|
||||
await this._updateTasks(updates, {
|
||||
isSkipDispatch: true,
|
||||
isIgnoreDBLock: options?.isIgnoreDBLock,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async unlinkIssueProviderFromArchiveTasks(
|
||||
unlinkIssueProviderFromArchiveTasks(
|
||||
issueProviderId: string,
|
||||
options?: { isIgnoreDBLock?: boolean },
|
||||
): Promise<void> {
|
||||
return this._runTaskArchiveMutation(
|
||||
() => this._unlinkIssueProviderFromArchiveTasks(issueProviderId),
|
||||
options?.isIgnoreDBLock,
|
||||
);
|
||||
}
|
||||
|
||||
private async _unlinkIssueProviderFromArchiveTasks(
|
||||
issueProviderId: string,
|
||||
): Promise<void> {
|
||||
const taskArchive = await this.load();
|
||||
|
||||
|
|
@ -466,14 +537,23 @@ export class TaskArchiveService {
|
|||
issuePoints: undefined,
|
||||
},
|
||||
}));
|
||||
await this.updateTasks(updates, {
|
||||
await this._updateTasks(updates, {
|
||||
isSkipDispatch: true,
|
||||
isIgnoreDBLock: options?.isIgnoreDBLock,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async roundTimeSpent({
|
||||
roundTimeSpent(params: {
|
||||
day: string;
|
||||
taskIds: string[];
|
||||
roundTo: RoundTimeOption;
|
||||
isRoundUp: boolean;
|
||||
projectId?: string | null;
|
||||
}): Promise<void> {
|
||||
return this._runTaskArchiveMutation(() => this._roundTimeSpent(params));
|
||||
}
|
||||
|
||||
private async _roundTimeSpent({
|
||||
day,
|
||||
taskIds,
|
||||
roundTo,
|
||||
|
|
|
|||
|
|
@ -457,6 +457,72 @@ describe('TaskService', () => {
|
|||
});
|
||||
|
||||
describe('moveToArchive', () => {
|
||||
it('should persist parent tasks before dispatching the archive action', async () => {
|
||||
const task = createMockTaskWithSubTasks(createMockTask('task-1'));
|
||||
const callOrder: string[] = [];
|
||||
archiveService.moveTasksToArchiveAndFlushArchiveIfDue.and.callFake(async () => {
|
||||
callOrder.push('persist-archive');
|
||||
});
|
||||
store.dispatch = jasmine
|
||||
.createSpy('dispatch')
|
||||
.and.callFake(() => callOrder.push('dispatch'));
|
||||
|
||||
await service.moveToArchive(task);
|
||||
|
||||
expect(callOrder).toEqual(['persist-archive', 'dispatch']);
|
||||
});
|
||||
|
||||
it('should coalesce concurrent archive requests for the same task', async () => {
|
||||
const task = createMockTaskWithSubTasks(createMockTask('task-1'));
|
||||
let finishPersistence: (() => void) | undefined;
|
||||
archiveService.moveTasksToArchiveAndFlushArchiveIfDue.and.returnValue(
|
||||
new Promise<void>((resolve) => {
|
||||
finishPersistence = resolve;
|
||||
}),
|
||||
);
|
||||
|
||||
const first = service.moveToArchive(task);
|
||||
const duplicate = service.moveToArchive(task);
|
||||
let duplicateResolved = false;
|
||||
void duplicate.then(() => {
|
||||
duplicateResolved = true;
|
||||
});
|
||||
|
||||
expect(archiveService.moveTasksToArchiveAndFlushArchiveIfDue).toHaveBeenCalledTimes(
|
||||
1,
|
||||
);
|
||||
expect(store.dispatch).not.toHaveBeenCalledWith(
|
||||
jasmine.objectContaining({ type: TaskSharedActions.moveToArchive.type }),
|
||||
);
|
||||
await Promise.resolve();
|
||||
expect(duplicateResolved).toBeFalse();
|
||||
|
||||
finishPersistence?.();
|
||||
await Promise.all([first, duplicate]);
|
||||
|
||||
expect(duplicateResolved).toBeTrue();
|
||||
expect(store.dispatch).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('should release the in-flight archive guard after persistence fails', async () => {
|
||||
const task = createMockTaskWithSubTasks(createMockTask('task-1'));
|
||||
archiveService.moveTasksToArchiveAndFlushArchiveIfDue.and.rejectWith(
|
||||
new Error('archive write failed'),
|
||||
);
|
||||
|
||||
await expectAsync(service.moveToArchive(task)).toBeRejectedWithError(
|
||||
'archive write failed',
|
||||
);
|
||||
|
||||
archiveService.moveTasksToArchiveAndFlushArchiveIfDue.and.resolveTo();
|
||||
await service.moveToArchive(task);
|
||||
|
||||
expect(archiveService.moveTasksToArchiveAndFlushArchiveIfDue).toHaveBeenCalledTimes(
|
||||
2,
|
||||
);
|
||||
expect(store.dispatch).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('should dispatch moveToArchive and call archive service for parent tasks', async () => {
|
||||
const task = createMockTaskWithSubTasks(
|
||||
createMockTask('task-1', { projectId: 'test-project' }),
|
||||
|
|
|
|||
|
|
@ -123,6 +123,7 @@ export class TaskService {
|
|||
private readonly _taskFocusService = inject(TaskFocusService);
|
||||
private readonly _deletedTaskIssueSidecar = inject(DeletedTaskIssueSidecarService);
|
||||
private readonly _timeBlockDeleteSidecar = inject(TimeBlockDeleteSidecarService);
|
||||
private readonly _archiveTaskPromisesById = new Map<string, Promise<void>>();
|
||||
|
||||
currentTaskId$: Observable<string | null> = this._store.pipe(
|
||||
select(selectCurrentTaskId),
|
||||
|
|
@ -947,15 +948,60 @@ export class TaskService {
|
|||
}
|
||||
}
|
||||
|
||||
if (parentTasks.length) {
|
||||
TaskLog.log('[TaskService] Dispatching moveToArchive action for parent tasks');
|
||||
const parentTasksToArchive: TaskWithSubTasks[] = [];
|
||||
const reservedTaskIds = new Set<string>();
|
||||
const existingArchivePromises = new Set<Promise<void>>();
|
||||
for (const task of parentTasks) {
|
||||
if (task.id) {
|
||||
const existingArchivePromise = this._archiveTaskPromisesById.get(task.id);
|
||||
if (existingArchivePromise) {
|
||||
TaskLog.log('[TaskService] Archive already in progress', { id: task.id });
|
||||
existingArchivePromises.add(existingArchivePromise);
|
||||
continue;
|
||||
}
|
||||
if (reservedTaskIds.has(task.id)) {
|
||||
continue;
|
||||
}
|
||||
reservedTaskIds.add(task.id);
|
||||
}
|
||||
parentTasksToArchive.push(task);
|
||||
}
|
||||
|
||||
if (parentTasksToArchive.length) {
|
||||
// Only move parent tasks to archive, never subtasks
|
||||
// Note: Full task payload required for sync - see docs/archive-operation-redesign.md
|
||||
this._store.dispatch(TaskSharedActions.moveToArchive({ tasks: parentTasks }));
|
||||
// Only archive parent tasks to prevent orphaned subtasks
|
||||
TaskLog.log('[TaskService] Calling archive service to persist tasks');
|
||||
await this._archiveService.moveTasksToArchiveAndFlushArchiveIfDue(parentTasks);
|
||||
TaskLog.log('[TaskService] Archive operation completed successfully');
|
||||
// Persist first: dispatch removes the tasks from NgRx and makes the captured
|
||||
// operation eligible for a full-state snapshot. If archive persistence were
|
||||
// still in flight, that snapshot could acknowledge the operation while
|
||||
// omitting its archived task data.
|
||||
const archivePromise = (async (): Promise<void> => {
|
||||
TaskLog.log('[TaskService] Calling archive service to persist tasks');
|
||||
await this._archiveService.moveTasksToArchiveAndFlushArchiveIfDue(
|
||||
parentTasksToArchive,
|
||||
);
|
||||
TaskLog.log('[TaskService] Dispatching moveToArchive action for parent tasks');
|
||||
this._store.dispatch(
|
||||
TaskSharedActions.moveToArchive({ tasks: parentTasksToArchive }),
|
||||
);
|
||||
TaskLog.log('[TaskService] Archive operation completed successfully');
|
||||
})();
|
||||
for (const taskId of reservedTaskIds) {
|
||||
this._archiveTaskPromisesById.set(taskId, archivePromise);
|
||||
}
|
||||
|
||||
try {
|
||||
await Promise.all([...existingArchivePromises, archivePromise]);
|
||||
} finally {
|
||||
for (const taskId of reservedTaskIds) {
|
||||
if (this._archiveTaskPromisesById.get(taskId) === archivePromise) {
|
||||
this._archiveTaskPromisesById.delete(taskId);
|
||||
}
|
||||
}
|
||||
}
|
||||
} else if (existingArchivePromises.size > 0) {
|
||||
// A duplicate caller observes the same success/failure and does not return
|
||||
// before the durable archive write plus NgRx removal have completed.
|
||||
await Promise.all(existingArchivePromises);
|
||||
} else {
|
||||
TaskLog.log('[TaskService] No parent tasks to archive');
|
||||
}
|
||||
|
|
|
|||
|
|
@ -49,6 +49,13 @@ import { InjectionToken } from '@angular/core';
|
|||
* IMPORTANT: These names are also used in tests - update tests if renaming.
|
||||
*/
|
||||
export const LOCK_NAMES = {
|
||||
/**
|
||||
* Serializes archive task read-modify-write cycles across tabs. This stays
|
||||
* separate from OPERATION_LOG because remote archive handlers already run
|
||||
* while holding that non-reentrant lock.
|
||||
*/
|
||||
TASK_ARCHIVE: 'sp_task_archive',
|
||||
|
||||
/**
|
||||
* Main operation log lock. Used for:
|
||||
* - Writing operations to IndexedDB
|
||||
|
|
@ -151,15 +158,6 @@ export const MAX_REJECTED_OPS_BEFORE_WARNING = 10;
|
|||
*/
|
||||
export const MAX_BATCH_OPERATIONS_SIZE = 50;
|
||||
|
||||
/**
|
||||
* Maximum age for pending operations before they are quarantined as failed (milliseconds).
|
||||
* If an operation has been pending for longer than this (e.g., due to data corruption
|
||||
* or repeated crashes), hydration still restores its reducer history and archive recovery
|
||||
* remains required.
|
||||
* Default: 24 hours - enough time for legitimate recovery scenarios
|
||||
*/
|
||||
export const PENDING_OPERATION_EXPIRY_MS = 24 * 60 * 60 * 1000;
|
||||
|
||||
/**
|
||||
* Threshold for warning about clock drift between client and server (milliseconds).
|
||||
* If the client's clock differs from the server by more than this amount,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue