Yield bounded catch-up batches behind per-user readiness queues
This commit is contained in:
@@ -8,9 +8,8 @@ import { HabitService } from "../habits/service";
|
|||||||
export function createAccountRoutes(db: AppDatabase, auth: ReturnType<typeof createAuth>, now: () => number) {
|
export function createAccountRoutes(db: AppDatabase, auth: ReturnType<typeof createAuth>, now: () => number) {
|
||||||
const app = new Hono<AuthEnv>();
|
const app = new Hono<AuthEnv>();
|
||||||
app.use("*", auth.requireAuth);
|
app.use("*", auth.requireAuth);
|
||||||
app.get("/export", c => {
|
app.get("/export", c => HabitService.withPrepared(db, c.get("user"), now, () => {
|
||||||
const user = c.get("user");
|
const user = c.get("user");
|
||||||
const service = new HabitService(db, user, now()); service.sync();
|
|
||||||
const data = db.transaction(tx => {
|
const data = db.transaction(tx => {
|
||||||
const owned = tx.select({ id: schema.habits.id }).from(schema.habits).where(eq(schema.habits.userId, user.id));
|
const owned = tx.select({ id: schema.habits.id }).from(schema.habits).where(eq(schema.habits.userId, user.id));
|
||||||
const days = tx.select({ id: schema.habitDays.id }).from(schema.habitDays).where(inArray(schema.habitDays.habitId, owned));
|
const days = tx.select({ id: schema.habitDays.id }).from(schema.habitDays).where(inArray(schema.habitDays.habitId, owned));
|
||||||
@@ -31,11 +30,12 @@ export function createAccountRoutes(db: AppDatabase, auth: ReturnType<typeof cre
|
|||||||
});
|
});
|
||||||
c.header("Content-Disposition", 'attachment; filename="mina-habits-export.json"');
|
c.header("Content-Disposition", 'attachment; filename="mina-habits-export.json"');
|
||||||
return c.json(data);
|
return c.json(data);
|
||||||
});
|
}));
|
||||||
app.delete("/", auth.requireSameOrigin, bodyLimit({ maxSize: 1024 }), async c => {
|
app.delete("/", auth.requireSameOrigin, bodyLimit({ maxSize: 1024 }), async c => {
|
||||||
const body = await c.req.json().catch(() => null);
|
const body = await c.req.json().catch(() => null);
|
||||||
if (body?.confirmation !== "DELETE") return c.json({ error: "Type DELETE to confirm account deletion." }, 422);
|
if (body?.confirmation !== "DELETE") return c.json({ error: "Type DELETE to confirm account deletion." }, 422);
|
||||||
const userId = c.get("user").id;
|
const userId = c.get("user").id;
|
||||||
|
return HabitService.withPrepared(db, c.get("user"), now, () => {
|
||||||
db.transaction(tx => {
|
db.transaction(tx => {
|
||||||
const owned = tx.select({ id: schema.habits.id }).from(schema.habits).where(eq(schema.habits.userId, userId));
|
const owned = tx.select({ id: schema.habits.id }).from(schema.habits).where(eq(schema.habits.userId, userId));
|
||||||
const days = tx.select({ id: schema.habitDays.id }).from(schema.habitDays).where(inArray(schema.habitDays.habitId, owned));
|
const days = tx.select({ id: schema.habitDays.id }).from(schema.habitDays).where(inArray(schema.habitDays.habitId, owned));
|
||||||
@@ -50,6 +50,7 @@ export function createAccountRoutes(db: AppDatabase, auth: ReturnType<typeof cre
|
|||||||
});
|
});
|
||||||
auth.clearSession(c);
|
auth.clearSession(c);
|
||||||
return c.body(null, 204);
|
return c.body(null, 204);
|
||||||
|
});
|
||||||
});
|
});
|
||||||
return app;
|
return app;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -57,3 +57,29 @@ test('unchanged dashboard and calendar reads write no rows after synchronization
|
|||||||
expect((f.sqlite.query('SELECT total_changes() AS n').get() as {n:number}).n).toBe(before);
|
expect((f.sqlite.query('SELECT total_changes() AS n').get() as {n:number}).n).toBe(before);
|
||||||
} finally {f.close();}
|
} finally {f.close();}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test('catch-up yields to other users and serializes same-user edits until fully ready', async () => {
|
||||||
|
const f = fixture();
|
||||||
|
try {
|
||||||
|
const habit = await f.json('/habits', 'POST', {name:'Many tasks', method:'tasks', tasks:Array.from({length:50}, (_,i)=>({name:`Task ${i}`}))}, 201);
|
||||||
|
f.setTime('2027-09-04T12:00:00Z');
|
||||||
|
let complete = false;
|
||||||
|
const dashboard = f.json('/today').then(value => {complete=true; return value;});
|
||||||
|
await new Promise(resolve=>setTimeout(resolve,0));
|
||||||
|
expect(complete).toBe(false);
|
||||||
|
// Another user's request can complete while Alice is still catching up.
|
||||||
|
expect((await f.request('/today','GET',undefined,'b')).status).toBe(200);
|
||||||
|
expect(complete).toBe(false);
|
||||||
|
// The actual date can advance while catch-up yields; readiness revalidates it.
|
||||||
|
f.setTime('2027-09-05T12:00:00Z');
|
||||||
|
const edit = f.json(`/habits/${habit.id}`, 'PATCH', {name:'Renamed'});
|
||||||
|
const [today, changed] = await Promise.all([dashboard, edit]);
|
||||||
|
expect(today.date).toBe('2027-09-05');
|
||||||
|
expect(today.habits[0].tasks).toHaveLength(50);
|
||||||
|
expect(changed.name).toBe('Renamed');
|
||||||
|
const through = f.sqlite.query('SELECT materialized_through AS date FROM habits WHERE id=?').get(habit.id) as {date:string};
|
||||||
|
expect(through.date).toBe('2027-09-05');
|
||||||
|
const day = await f.json(`/habits/${habit.id}/days/2026-09-05`);
|
||||||
|
expect(day.tasks.every((task:any)=>task.expiredAt!==null && task.closedAt!==null)).toBe(true);
|
||||||
|
} finally {f.close();}
|
||||||
|
});
|
||||||
|
|||||||
@@ -41,8 +41,9 @@ export function createHabitRoutes(db: AppDatabase, auth: ReturnType<typeof creat
|
|||||||
try { payload.value = await c.req.json(); } catch { payload.malformed = true; }
|
try { payload.value = await c.req.json(); } catch { payload.malformed = true; }
|
||||||
}
|
}
|
||||||
c.set('payload', payload);
|
c.set('payload', payload);
|
||||||
const service = new HabitService(db, c.get('user'), now());
|
await HabitService.withPrepared(db, c.get('user'), now, async service => {
|
||||||
service.sync(); c.set('service', service); await next();
|
c.set('service', service); await next();
|
||||||
|
});
|
||||||
}); if (response) c.res = response; };
|
}); if (response) c.res = response; };
|
||||||
if (!['GET', 'HEAD', 'OPTIONS'].includes(c.req.method)) {
|
if (!['GET', 'HEAD', 'OPTIONS'].includes(c.req.method)) {
|
||||||
const response = await origin(c, proceed); if (response) c.res = response;
|
const response = await origin(c, proceed); if (response) c.res = response;
|
||||||
|
|||||||
@@ -13,7 +13,45 @@ export type Habit = typeof habits.$inferSelect;
|
|||||||
export type Revision = typeof habitRevisions.$inferSelect;
|
export type Revision = typeof habitRevisions.$inferSelect;
|
||||||
export type HabitDay = typeof habitDays.$inferSelect;
|
export type HabitDay = typeof habitDays.$inferSelect;
|
||||||
|
|
||||||
|
// Each database has its own per-user readiness queue. No transaction spans a yield.
|
||||||
|
const preparationQueues = new WeakMap<AppDatabase, Map<string, Promise<void>>>();
|
||||||
export class HabitService {
|
export class HabitService {
|
||||||
|
static async withPrepared<T>(db: AppDatabase, user: PublicUser, now: () => number, run: (service: HabitService) => T | Promise<T>): Promise<T> {
|
||||||
|
let queues = preparationQueues.get(db);
|
||||||
|
if (!queues) { queues = new Map(); preparationQueues.set(db, queues); }
|
||||||
|
const previous = queues.get(user.id);
|
||||||
|
let release!: () => void;
|
||||||
|
const ticket = new Promise<void>(resolve => { release = resolve; });
|
||||||
|
queues.set(user.id, ticket);
|
||||||
|
if (previous) await previous;
|
||||||
|
try {
|
||||||
|
let sliceStarted = performance.now();
|
||||||
|
while (true) {
|
||||||
|
// Re-read timezone, date and owned records after every scheduling yield.
|
||||||
|
const service = new HabitService(db, user, now());
|
||||||
|
const habit = service.pending()[0];
|
||||||
|
if (!habit) {
|
||||||
|
service.sync();
|
||||||
|
return await run(service);
|
||||||
|
}
|
||||||
|
const start = habit.materializedThrough ? addDays(habit.materializedThrough, 1) : habit.createdDate;
|
||||||
|
const through = [addDays(start, 7), service.today].sort()[0]!;
|
||||||
|
db.transaction(() => service.materialize(habit, through));
|
||||||
|
if (performance.now() - sliceStarted >= 8) {
|
||||||
|
await new Promise<void>(resolve => setTimeout(resolve, 0));
|
||||||
|
sliceStarted = performance.now();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} finally {
|
||||||
|
release();
|
||||||
|
if (queues.get(user.id) === ticket) queues.delete(user.id);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
/** Ready read snapshot. Use withPrepared for read-modify-write handlers. */
|
||||||
|
static prepare(db: AppDatabase, user: PublicUser, now: () => number) {
|
||||||
|
return HabitService.withPrepared(db, user, now, service => service);
|
||||||
|
}
|
||||||
|
|
||||||
readonly today: string;
|
readonly today: string;
|
||||||
private ownedCache = new Map<string, Habit>();
|
private ownedCache = new Map<string, Habit>();
|
||||||
private revisionCache = new Map<string, Revision[]>();
|
private revisionCache = new Map<string, Revision[]>();
|
||||||
@@ -25,7 +63,9 @@ export class HabitService {
|
|||||||
private countCache = new Map<string, Map<string, { value: number; carriedFrom: string | null }>>();
|
private countCache = new Map<string, Map<string, { value: number; carriedFrom: string | null }>>();
|
||||||
constructor(readonly db: AppDatabase, readonly user: PublicUser, readonly now: number) {
|
constructor(readonly db: AppDatabase, readonly user: PublicUser, readonly now: number) {
|
||||||
// A different request may update the timezone while this request's body streams in.
|
// A different request may update the timezone while this request's body streams in.
|
||||||
const timezone = db.select({ timezone: users.timezone }).from(users).where(eq(users.id, user.id)).get()!.timezone;
|
const currentUser = db.select({ timezone: users.timezone }).from(users).where(eq(users.id, user.id)).get();
|
||||||
|
if (!currentUser) throw missing();
|
||||||
|
const timezone = currentUser.timezone;
|
||||||
this.user = { ...user, timezone };
|
this.user = { ...user, timezone };
|
||||||
// Travel west must never reopen or rewrite earlier requirements.
|
// Travel west must never reopen or rewrite earlier requirements.
|
||||||
const latest = db.select({ date: habits.materializedThrough }).from(habits).where(eq(habits.userId, user.id)).orderBy(desc(habits.materializedThrough)).get()?.date;
|
const latest = db.select({ date: habits.materializedThrough }).from(habits).where(eq(habits.userId, user.id)).orderBy(desc(habits.materializedThrough)).get()?.date;
|
||||||
@@ -221,11 +261,12 @@ export class HabitService {
|
|||||||
.innerJoin(habitDays, eq(taskOccurrences.dayId, habitDays.id))
|
.innerJoin(habitDays, eq(taskOccurrences.dayId, habitDays.id))
|
||||||
.where(and(eq(habitDays.habitId, habit.id), eq(habitDays.date, date)))
|
.where(and(eq(habitDays.habitId, habit.id), eq(habitDays.date, date)))
|
||||||
.orderBy(desc(habitDays.id)).all().map(row => row.occurrence) : [];
|
.orderBy(desc(habitDays.id)).all().map(row => row.occurrence) : [];
|
||||||
for (const task of revision.config.tasks.filter(t => scheduled(t.schedule, date))) {
|
const occurrences = revision.config.tasks.filter(t => scheduled(t.schedule, date)).map(task => {
|
||||||
const old = previous.find(t => t.taskId === task.id);
|
const old = previous.find(t => t.taskId === task.id);
|
||||||
this.db.insert(taskOccurrences).values({ id: crypto.randomUUID(), dayId: day.id, taskId: task.id, name: task.name,
|
return { id: crypto.randomUUID(), dayId: day.id, taskId: task.id, name: task.name,
|
||||||
done: old?.done ?? false, expiredAt: old ? old.expiredAt : day.endsAt <= this.now ? day.endsAt : null, closedAt: old?.closedAt ?? (day.endsAt <= this.now ? day.endsAt : null), updatedAt: old?.updatedAt ?? this.now }).run();
|
done: old?.done ?? false, expiredAt: old ? old.expiredAt : day.endsAt <= this.now ? day.endsAt : null, closedAt: old?.closedAt ?? (day.endsAt <= this.now ? day.endsAt : null), updatedAt: old?.updatedAt ?? this.now };
|
||||||
}
|
});
|
||||||
|
if (occurrences.length) this.db.insert(taskOccurrences).values(occurrences).run();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
date = addDays(date, 1);
|
date = addDays(date, 1);
|
||||||
@@ -240,7 +281,7 @@ export class HabitService {
|
|||||||
const rows = select().where(and(eq(habitDays.habitId, id), gte(habitDays.date, range.from), lte(habitDays.date, range.to), latest)).orderBy(asc(habitDays.date)).all();
|
const rows = select().where(and(eq(habitDays.habitId, id), gte(habitDays.date, range.from), lte(habitDays.date, range.to), latest)).orderBy(asc(habitDays.date)).all();
|
||||||
// Only carry-enabled ranges need a prefix. Walk backwards in bounded pages
|
// Only carry-enabled ranges need a prefix. Walk backwards in bounded pages
|
||||||
// until a logged value or requirements reset establishes the starting state.
|
// until a logged value or requirements reset establishes the starting state.
|
||||||
const first = rows[0];
|
const first = rows.find(({day, config}) => config.method !== 'count' || config.archived || scheduled(config.schedule, day.date));
|
||||||
if (first?.config.method === 'count' && first.config.carryPartialProgress && !first.day.countSet) {
|
if (first?.config.method === 'count' && first.config.carryPartialProgress && !first.day.countSet) {
|
||||||
let before = range.from;
|
let before = range.from;
|
||||||
const prefix: typeof rows = [];
|
const prefix: typeof rows = [];
|
||||||
|
|||||||
Reference in New Issue
Block a user