fix(contribution): LEVEL_OVERFLOW 回收任务,修复已解锁层级的溢出记录无法被回收的 bug
当下级认种时上级 unlocked_level_depth 不足,层级奖励进入 LEVEL_OVERFLOW(PENDING)。 上级后续解锁到足够层级后,现有 backfill 因条件 expectedLevel > currentLevel 为 false 而跳过,导致 PENDING 记录永远无法被回收。新增独立调度任务每10分钟扫描并回收。 Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
parent
ca4e5393be
commit
b1607666a0
|
|
@ -253,6 +253,35 @@ export class ContributionScheduler implements OnModuleInit {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 每10分钟回收已解锁层级的 LEVEL_OVERFLOW 记录
|
||||||
|
* 处理场景:下级认种时上级 unlocked_level_depth 不足导致 overflow,
|
||||||
|
* 上级后续解锁后这些 PENDING 记录需要被回收
|
||||||
|
*/
|
||||||
|
@Cron('*/10 * * * *')
|
||||||
|
async processLevelOverflowReclaim(): Promise<void> {
|
||||||
|
if (!this.isCdcReady()) {
|
||||||
|
this.logger.debug('[CDC-Gate] processLevelOverflowReclaim skipped: CDC initial sync not yet completed');
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
const lockValue = await this.redis.acquireLock(`${this.LOCK_KEY}:overflow-reclaim`, 540); // 9分钟锁
|
||||||
|
if (!lockValue) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
try {
|
||||||
|
const reclaimed = await this.bonusClaimService.reclaimLevelOverflows();
|
||||||
|
if (reclaimed > 0) {
|
||||||
|
this.logger.log(`Level overflow reclaim: ${reclaimed} records reclaimed`);
|
||||||
|
}
|
||||||
|
} catch (error) {
|
||||||
|
this.logger.error('Failed to process level overflow reclaim', error);
|
||||||
|
} finally {
|
||||||
|
await this.redis.releaseLock(`${this.LOCK_KEY}:overflow-reclaim`, lockValue);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 每10分钟扫描并补发未完全解锁的贡献值
|
* 每10分钟扫描并补发未完全解锁的贡献值
|
||||||
* 处理因下级先于上级认种导致的层级/奖励档位未能及时分配的情况
|
* 处理因下级先于上级认种导致的层级/奖励档位未能及时分配的情况
|
||||||
|
|
|
||||||
|
|
@ -273,6 +273,66 @@ export class BonusClaimService {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ========== LEVEL_OVERFLOW 回收逻辑 ==========
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 回收已解锁层级的 LEVEL_OVERFLOW 记录
|
||||||
|
* 处理场景:下级认种时上级 unlocked_level_depth 不足,产生 LEVEL_OVERFLOW;
|
||||||
|
* 后续上级解锁到足够层级后,这些 PENDING 记录需要被回收分配
|
||||||
|
* @param limit 每次扫描的最大账户数
|
||||||
|
* @returns 回收的记录总数
|
||||||
|
*/
|
||||||
|
async reclaimLevelOverflows(limit: number = 100): Promise<number> {
|
||||||
|
// 1. 查找有 PENDING LEVEL_OVERFLOW 记录的账户
|
||||||
|
const accountSequences = await this.unallocatedContributionRepository
|
||||||
|
.findAccountSequencesWithPendingLevelOverflow(limit);
|
||||||
|
|
||||||
|
if (accountSequences.length === 0) {
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
this.logger.log(`[OverflowReclaim] Found ${accountSequences.length} accounts with pending LEVEL_OVERFLOW`);
|
||||||
|
|
||||||
|
let totalReclaimed = 0;
|
||||||
|
let errorCount = 0;
|
||||||
|
|
||||||
|
for (const accountSequence of accountSequences) {
|
||||||
|
try {
|
||||||
|
const account = await this.contributionAccountRepository.findByAccountSequence(accountSequence);
|
||||||
|
if (!account || account.unlockedLevelDepth === 0) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
// 只回收已解锁层级范围内的 overflow
|
||||||
|
await this.unitOfWork.executeInTransaction(async () => {
|
||||||
|
const claimed = await this.claimLevelContributions(
|
||||||
|
accountSequence,
|
||||||
|
1,
|
||||||
|
account.unlockedLevelDepth,
|
||||||
|
);
|
||||||
|
|
||||||
|
if (claimed > 0) {
|
||||||
|
totalReclaimed += claimed;
|
||||||
|
// 重新读取账户(claimLevelContributions 已更新余额),发布更新事件
|
||||||
|
const updatedAccount = await this.contributionAccountRepository
|
||||||
|
.findByAccountSequence(accountSequence);
|
||||||
|
if (updatedAccount) {
|
||||||
|
await this.publishContributionAccountUpdatedEvent(updatedAccount);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
} catch (error) {
|
||||||
|
errorCount++;
|
||||||
|
this.logger.error(`[OverflowReclaim] Failed for account ${accountSequence}`, error);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
this.logger.log(
|
||||||
|
`[OverflowReclaim] Completed: ${totalReclaimed} records reclaimed, ${errorCount} errors`,
|
||||||
|
);
|
||||||
|
return totalReclaimed;
|
||||||
|
}
|
||||||
|
|
||||||
// ========== 定时任务补发逻辑 ==========
|
// ========== 定时任务补发逻辑 ==========
|
||||||
|
|
||||||
private readonly domainCalculator = new ContributionCalculatorService();
|
private readonly domainCalculator = new ContributionCalculatorService();
|
||||||
|
|
|
||||||
|
|
@ -280,6 +280,25 @@ export class UnallocatedContributionRepository {
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 查询有待回收 LEVEL_OVERFLOW 记录的用户账号列表
|
||||||
|
* 用于定时任务扫描:当用户已解锁到足够层级,但之前的 overflow 尚未回收时
|
||||||
|
*/
|
||||||
|
async findAccountSequencesWithPendingLevelOverflow(limit: number): Promise<string[]> {
|
||||||
|
const records = await this.client.unallocatedContribution.findMany({
|
||||||
|
where: {
|
||||||
|
unallocType: 'LEVEL_OVERFLOW',
|
||||||
|
status: 'PENDING',
|
||||||
|
wouldBeAccountSequence: { not: null },
|
||||||
|
},
|
||||||
|
select: { wouldBeAccountSequence: true },
|
||||||
|
distinct: ['wouldBeAccountSequence'],
|
||||||
|
take: limit,
|
||||||
|
});
|
||||||
|
|
||||||
|
return records.map((r) => r.wouldBeAccountSequence!);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 获取分档位的未分配奖励统计
|
* 获取分档位的未分配奖励统计
|
||||||
*/
|
*/
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue