diff --git a/laf-cloud/docs/pass-check-settlement.txt b/laf-cloud/docs/pass-check-settlement.txt index 350ee1e..bf01b71 100644 --- a/laf-cloud/docs/pass-check-settlement.txt +++ b/laf-cloud/docs/pass-check-settlement.txt @@ -12,20 +12,22 @@ passCheckJobs 不开放 HTTP 方法；需要现有 PASSCHECKTIME_ID 环境配置。 每分钟仅用于检查周期和继续未完成的批次；本期扫描完成后不再读取用户表。 3. 再发布配套客户端。旧客户端的过期进度写入将被拒绝，不能继续按旧方式领上期奖励。 -4. 不清空用户历史。快照存入 users/usersAd 的 passSettlements， - passSettlementRevision 用于 CAS。后台每期每个用户集合只完成一轮分页扫描， - 每次触发各处理最多100条；进度保存在 passCheckJobCursors。 +4. 不清空用户历史。后台仅处理 users，快照存入其 passSettlements， + passSettlementRevision 用于 CAS。每期只完成一轮分页扫描， + 每次触发处理最多100条；进度保存在 passCheckJobCursors。 请按用户规模评估扫描延迟；玩家请求时同样惰性补建快照，不必等待后台扫到自己。 按期扫描与旧任务升级 - 当前全局周期起点同时是上一期结束点，以该毫秒时间戳标识扫描期。 即使在线接口先推进了全局周期，后台仍会对比自己的 periodEnd，启动新一期扫描。 -- 新进度键为 season:users、season:usersAd；保存 periodEnd、after、completed、 +- 进度键为 season:users；保存 periodEnd、after、completed、 completedAt、revision、leaseUntil。仅使用已有集合的 _id 索引，无需手动初始化。 - 保留旧 users/usersAd 游标，但不再读取它们；升级后的首次运行从头补扫一次历史存档。 + 保留旧 users/usersAd 游标，但不再读取它们；没有 season:users 时从头补扫一次历史存档。 + 已有 season:users 则继续使用原进度。任务不访问 usersAd 集合，也不读取、更新或删除 + 历史 season:usersAd 进度记录。 此后只有全局周期推进才开启新一轮；配置的起点还在未来时不扫描。 - 每批成功后推进游标，不足100条时标记完成；恰好整批时下一次空页确认完成。 - 两个集合分别完成。全部完成后每次触发仅读取周期配置和两个进度文档，不扫描用户， + 完成后每次触发仅读取周期配置和 season:users 进度文档，不扫描用户， 不反复更新游标；云函数每分钟的触发次数本身未减少。 - 每批取得5分钟租约，并用 revision 比较更新防止重叠执行或旧执行覆盖新进度。 批次失败保留起始游标、释放租约，下次重试；进程直接退出时租约到期后恢复。 @@ -62,9 +64,10 @@ claimFlags 仅更新实际领取的 free/passCheck 索引为0，不修改冻结 需人工退款处理，不自动退款。 验证 -Node 24：node --test laf-cloud/tests/pass-check-settlement.test.mjs（15项通过）。 +Node 24：node --test laf-cloud/tests/pass-check-settlement.test.mjs（16项通过）。 覆盖完成后零用户扫描/零写入、分页、新周期、空集合/整批边界、失败重试、旧游标升级、 -跨期续跑、租约超时接管与旧执行防回退、在线补建、未来周期配置；保留原发奖幂等测试。 +跨期续跑、租约超时接管与旧执行防回退、在线补建、未来周期配置、完全不访问 usersAd +及其旧进度记录；保留原发奖幂等测试。 配套客户端 tools/test-battle-pass-settlement.cjs 覆盖奖励表、UI关闭=领取、 失败重试、确认弹窗、优先队列、多期顺序和支付幂等。 现有 rookie-gift.test.mjs 在导入阶段失败：goldMiner/config 缺少 paymentProductId diff --git a/laf-cloud/functions/passCheckJobs.ts b/laf-cloud/functions/passCheckJobs.ts index a13e672..be68e10 100644 --- a/laf-cloud/functions/passCheckJobs.ts +++ b/laf-cloud/functions/passCheckJobs.ts @@ -12,47 +12,45 @@ export default async function () { if (periodEnd > Date.now()) return { count: 0 }; const db: any = cloud.mongo.db; let count = 0; - for (const table of ['users', 'usersAd']) { - const cursors = db.collection('passCheckJobCursors'); - // Separate keys intentionally ignore legacy circular cursors: the first - // run backfills from the beginning once, without deleting old job state. - const key = { _id: 'season:' + table }; - let cursor = await cursors.findOne(key); - if (!cursor) { - try { - await cursors.updateOne(key, { $setOnInsert: { - periodEnd, after: null, completed: false, revision: 0, leaseUntil: 0 - } }, { upsert: true }); - } catch (error: any) { if (error.code !== 11000) throw error; } - cursor = await cursors.findOne(key); - } - if (cursor.completed && cursor.periodEnd >= periodEnd) continue; - const now = Date.now(); - if (cursor.leaseUntil > now) continue; - const next = cursor.completed - ? { periodEnd, after: null, completed: false, completedAt: null } - : { periodEnd: cursor.periodEnd, after: cursor.after, completed: false }; - const locked = await cursors.updateOne({ ...key, revision: cursor.revision, leaseUntil: { $lte: now } }, { - $set: { ...next, leaseUntil: now + LEASE_MS }, $inc: { revision: 1 } - }); - if (!locked.modifiedCount) continue; - const owned = { ...key, revision: cursor.revision + 1 }; + const cursors = db.collection('passCheckJobCursors'); + // This key intentionally ignores the legacy circular cursor: the first + // run backfills from the beginning once, without deleting old job state. + const key = { _id: 'season:users' }; + let cursor = await cursors.findOne(key); + if (!cursor) { try { - const rows = await db.collection(table).find(next.after != null ? { _id: { $gt: next.after } } : {}) - .sort({ _id: 1 }).limit(BATCH_SIZE).toArray(); - for (const user of rows) { await freezeUser(user, table); count++; } - const completed = rows.length < BATCH_SIZE; - await cursors.updateOne(owned, { - $set: { after: rows.length ? rows[rows.length - 1]._id : next.after, completed, - completedAt: completed ? Date.now() : null, leaseUntil: 0 }, - $inc: { revision: 1 } - }); - } catch (error) { - // Keep this batch's starting cursor. Freezing is idempotent on retry; - // revision fencing prevents an expired worker overwriting its successor. - await cursors.updateOne(owned, { $set: { leaseUntil: 0 }, $inc: { revision: 1 } }); - throw error; - } + await cursors.updateOne(key, { $setOnInsert: { + periodEnd, after: null, completed: false, revision: 0, leaseUntil: 0 + } }, { upsert: true }); + } catch (error: any) { if (error.code !== 11000) throw error; } + cursor = await cursors.findOne(key); + } + if (cursor.completed && cursor.periodEnd >= periodEnd) return { count }; + const now = Date.now(); + if (cursor.leaseUntil > now) return { count }; + const next = cursor.completed + ? { periodEnd, after: null, completed: false, completedAt: null } + : { periodEnd: cursor.periodEnd, after: cursor.after, completed: false }; + const locked = await cursors.updateOne({ ...key, revision: cursor.revision, leaseUntil: { $lte: now } }, { + $set: { ...next, leaseUntil: now + LEASE_MS }, $inc: { revision: 1 } + }); + if (!locked.modifiedCount) return { count }; + const owned = { ...key, revision: cursor.revision + 1 }; + try { + const rows = await db.collection('users').find(next.after != null ? { _id: { $gt: next.after } } : {}) + .sort({ _id: 1 }).limit(BATCH_SIZE).toArray(); + for (const user of rows) { await freezeUser(user, 'users'); count++; } + const completed = rows.length < BATCH_SIZE; + await cursors.updateOne(owned, { + $set: { after: rows.length ? rows[rows.length - 1]._id : next.after, completed, + completedAt: completed ? Date.now() : null, leaseUntil: 0 }, + $inc: { revision: 1 } + }); + } catch (error) { + // Keep this batch's starting cursor. Freezing is idempotent on retry; + // revision fencing prevents an expired worker overwriting its successor. + await cursors.updateOne(owned, { $set: { leaseUntil: 0 }, $inc: { revision: 1 } }); + throw error; } return { count }; } diff --git a/laf-cloud/tests/pass-check-settlement.test.mjs b/laf-cloud/tests/pass-check-settlement.test.mjs index 8ff2731..c59ddab 100644 --- a/laf-cloud/tests/pass-check-settlement.test.mjs +++ b/laf-cloud/tests/pass-check-settlement.test.mjs @@ -4,7 +4,7 @@ import { registerHooks } from 'node:module'; const NOW = 1800000000000, END = NOW - 1000, PERIOD = 30 * 86400000; let now = NOW, drop = false, failFreeze = '', beforeFind = null; -const scans = [], writes = []; +const reads = [], scans = [], writes = []; const tables = { users: [], usersAd: [], order: [], idcount: [], passCheckJobCursors: [] }; const clone = v => v == null ? v : structuredClone(v); const get = (r, k) => k.split('.').reduce((v, key) => v?.[key], r); @@ -16,7 +16,10 @@ const match = (r, q) => Object.entries(q).every(([k, v]) => { return JSON.stringify(get(r, k)) === JSON.stringify(v); }); const collection = name => ({ - async findOne(q) { return clone(tables[name].find(r => match(r, q))); }, + async findOne(q) { + reads.push({ name, q: clone(q) }); + return clone(tables[name].find(r => match(r, q))); + }, find(q) { let limit = Infinity; const query = { @@ -71,7 +74,7 @@ Date.now = () => now; test.after(() => { Date.now = realNow; }); function reset() { now = NOW; drop = false; failFreeze = ''; beforeFind = null; - scans.length = 0; writes.length = 0; + reads.length = 0; scans.length = 0; writes.length = 0; for (const key of Object.keys(tables)) tables[key] = []; process.env.PASSCHECKTIME_ID = 'clock'; tables.idcount.push({ _id: 'clock', passcheckTime: END - PERIOD }); @@ -144,16 +147,18 @@ function populate(count) { tables.users = Array.from({ length: count }, (_, i) => ({ ...clone(original), _id: 'u' + String(i).padStart(4, '0') })); } -test('job completes each collection once and idle ticks never scan users or write cursors', async () => { +test('job scans only users once and idle ticks never scan users or write cursors', async () => { const u = reset(); tables.usersAd.push({ ...clone(u), _id: 'ad' }); - assert.equal((await runJob()).count, 2); - for (const table of ['users', 'usersAd']) { - assert.equal(jobCursor(table).completed, true); - assert.equal(jobCursor(table).periodEnd, END); - assert.ok(tables[table][0].passSettlements[END]); - assert.equal(tables[table][0].coinAmount, 10); - } + const adBefore = clone(tables.usersAd); + assert.equal((await runJob()).count, 1); + assert.equal(jobCursor('users').completed, true); + assert.equal(jobCursor('users').periodEnd, END); + assert.ok(u.passSettlements[END]); + assert.equal(u.coinAmount, 10); + assert.equal(jobCursor('usersAd'), undefined); + assert.deepEqual(tables.usersAd, adBefore); + assert.ok([...reads, ...scans, ...writes].every(r => r.name !== 'usersAd' && r.q._id !== 'season:usersAd')); scans.length = 0; writes.length = 0; for (let i = 0; i < 5; i++) { now += 60000; assert.equal((await runJob()).count, 0); } assert.deepEqual(scans, []); @@ -182,11 +187,11 @@ test('job resumes batches without rewinding and only opens a new sweep at the ne assert.equal(u.coinAmount, 10); }); -test('empty collections and exact full batches terminate instead of restarting', async () => { +test('exact full batches terminate instead of restarting', async () => { populate(100); assert.equal((await runJob()).count, 100); assert.equal(jobCursor('users').completed, false); - assert.equal(jobCursor('usersAd').completed, true); + assert.equal(jobCursor('usersAd'), undefined); assert.equal((await runJob()).count, 0); assert.equal(jobCursor('users').completed, true); scans.length = 0; @@ -218,6 +223,22 @@ test('legacy circular cursors are preserved but never used to skip the initial b assert.deepEqual(tables.passCheckJobCursors.find(r => r._id === 'users'), { _id: 'users', after: 'z' }); }); +test('obsolete usersAd job state is never read or updated, including across periods', async () => { + const u = reset(); + tables.usersAd.push({ ...clone(u), _id: 'ad' }); + const oldCursor = { _id: 'season:usersAd', periodEnd: END - PERIOD, after: null, + completed: false, revision: 2, leaseUntil: 0 }; + tables.passCheckJobCursors.push(clone(oldCursor)); + const adBefore = clone(tables.usersAd); + await runJob(); + await runJob(); + now = END + PERIOD; + await runJob(); + assert.deepEqual(jobCursor('usersAd'), oldCursor); + assert.deepEqual(tables.usersAd, adBefore); + assert.ok([...reads, ...scans, ...writes].every(r => r.name !== 'usersAd' && r.q._id !== 'season:usersAd')); +}); + test('cross-period recovery finishes the old sweep before starting the latest period', async () => { populate(101); await runJob();