fix Redis active-plan race between /v2/vibe/next and /feedback (#221)
Per-user SET NX PX lock with Lua CAS release around the getActivePlan->mutate->setActivePlan span so concurrent prefetch + feedback requests serialize instead of losing one side's write. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
import { FastifyInstance } from 'fastify';
|
||||
import { createClient, RedisClientType } from 'redis';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { DbService } from '../services/db.service.js';
|
||||
import { SessionDirector } from '../services/session-director.service.js';
|
||||
import { Candidate } from '../services/generators.service.js';
|
||||
@@ -39,6 +40,34 @@ export default async function v2Routes(fastify: FastifyInstance, options: { dbSe
|
||||
await redisClient.del(planKey(userId));
|
||||
}
|
||||
|
||||
// ponytail: per-user SET NX PX lock around the getActivePlan->mutate->setActivePlan
|
||||
// span, released via a CAS Lua script so a slow holder can't delete a lock a later
|
||||
// request already owns. Ceiling: this serializes /next and /feedback per-user only
|
||||
// (fine — plans aren't shared across users); if plan mutation logic ever needs to
|
||||
// span multiple keys/services atomically, swap this for a WATCH/MULTI transaction
|
||||
// or move the plan into a single Lua script instead of an app-level lock.
|
||||
const RELEASE_LOCK_LUA = `if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end`;
|
||||
|
||||
async function withPlanLock<T>(userId: string, fn: () => Promise<T>): Promise<T> {
|
||||
const lockKey = `v2:planlock:${userId}`;
|
||||
const token = randomUUID();
|
||||
const deadline = Date.now() + 5000;
|
||||
let acquired = false;
|
||||
while (Date.now() < deadline) {
|
||||
const res = await redisClient.set(lockKey, token, { NX: true, PX: 3000 });
|
||||
if (res) { acquired = true; break; }
|
||||
await new Promise((r) => setTimeout(r, 20 + Math.random() * 30));
|
||||
}
|
||||
if (!acquired) {
|
||||
throw new Error('Timed out waiting for active-plan lock');
|
||||
}
|
||||
try {
|
||||
return await fn();
|
||||
} finally {
|
||||
await redisClient.eval(RELEASE_LOCK_LUA, { keys: [lockKey], arguments: [token] });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* POST /api/v2/vibe/start — start a v2 session
|
||||
* Body: { seedTrackId? }
|
||||
@@ -61,25 +90,34 @@ export default async function v2Routes(fastify: FastifyInstance, options: { dbSe
|
||||
*/
|
||||
fastify.get('/v2/vibe/next', async (request, reply) => {
|
||||
const userId = (request.headers['x-user-id'] as string) || '00000000-0000-0000-0000-000000000000';
|
||||
const active = await getActivePlan(userId);
|
||||
|
||||
if (!active || active.plan.length === 0) {
|
||||
const result = await withPlanLock(userId, async () => {
|
||||
const active = await getActivePlan(userId);
|
||||
|
||||
if (!active || active.plan.length === 0) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const next = active.plan.shift()!;
|
||||
// Enrich with track details
|
||||
const track = await dbService.getTrackById(next.trackId);
|
||||
|
||||
// Replan if running low
|
||||
if (active.plan.length < 5) {
|
||||
const refill = await director.replan(userId, active.sessionId, active.plan, [next.trackId], active.seedTrackId ?? undefined);
|
||||
active.plan = refill;
|
||||
}
|
||||
|
||||
await setActivePlan(userId, active);
|
||||
|
||||
return { track, explanation: next.explanation, planRemaining: active.plan.length };
|
||||
});
|
||||
|
||||
if (!result) {
|
||||
return reply.code(404).send({ error: 'No active plan. POST /api/v2/vibe/start first.' });
|
||||
}
|
||||
|
||||
const next = active.plan.shift()!;
|
||||
// Enrich with track details
|
||||
const track = await dbService.getTrackById(next.trackId);
|
||||
|
||||
// Replan if running low
|
||||
if (active.plan.length < 5) {
|
||||
const refill = await director.replan(userId, active.sessionId, active.plan, [next.trackId], active.seedTrackId ?? undefined);
|
||||
active.plan = refill;
|
||||
}
|
||||
|
||||
await setActivePlan(userId, active);
|
||||
|
||||
return reply.send({ track, explanation: next.explanation, planRemaining: active.plan.length });
|
||||
return reply.send(result);
|
||||
});
|
||||
|
||||
/**
|
||||
@@ -107,15 +145,17 @@ export default async function v2Routes(fastify: FastifyInstance, options: { dbSe
|
||||
}
|
||||
|
||||
// Replan the session
|
||||
const active = await getActivePlan(userId);
|
||||
if (active) {
|
||||
const planRemaining = await withPlanLock(userId, async () => {
|
||||
const active = await getActivePlan(userId);
|
||||
if (!active) return 0;
|
||||
const playedTrackIds = [trackId];
|
||||
const refill = await director.replan(userId, active.sessionId, active.plan, playedTrackIds, active.seedTrackId ?? undefined);
|
||||
active.plan = refill;
|
||||
await setActivePlan(userId, active);
|
||||
}
|
||||
return active.plan.length;
|
||||
});
|
||||
|
||||
return reply.send({ status: 'ok', planRemaining: active?.plan.length ?? 0 });
|
||||
return reply.send({ status: 'ok', planRemaining });
|
||||
});
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user