fix(playback): 登录合并前预写设备缓冲,迟到条目不得复活已删记录

- 新增 flushIdentity:把某身份缓冲里未落库的进度立即写进权威表;
  mergeDeviceIntoUser 先调用它再扫库,否则登录前 2 秒内的上报会被漏掉,
  稍后落库还会生成一条登录用户看不到的匿名孤儿行;合并完清掉设备侧缓冲。
- planOne 增加判定:逻辑删除会刷新 updated_at,上报时间不晚于它的条目直接丢弃,
  防止删除瞬间已 drain 的在途引用把 deleted 写回 0。
- 去掉 save(Long userId, ...) 上多余的 @Transactional(缓冲路径白开事务)。
This commit is contained in:
Zhou Weidong authored and trae-solo-agent committed 2026-10-28 19:02:32 +08:00
1 parent 025fbe73f0
commit a71d2701d1
1 file changed
+44 -1
@@ -85,7 +85,6 @@ public class WatchProgressService {
.expireAfterWrite(Duration.ofMinutes(5)) .expireAfterWrite(Duration.ofMinutes(5))
.build(); .build();
@Transactional(rollbackFor = Exception.class)
public ProgressRespVO save(Long userId, ProgressSaveReqVO req) { public ProgressRespVO save(Long userId, ProgressSaveReqVO req) {
return save(new InteractionIdentity(userId, ""), req); return save(new InteractionIdentity(userId, ""), req);
} }
@@ -460,6 +459,12 @@ public class WatchProgressService {
return; return;
} }
boolean hidden = row.getDeleted() != null && row.getDeleted() != 0; boolean hidden = row.getDeleted() != null && row.getDeleted() != 0;
long rowTs = toMillis(row.getUpdatedAt());
if (hidden && rowTs > 0 && state.ts() <= rowTs) {
// 逻辑删除会刷新 updated_at:上报时间早于它的条目是删除前排队的迟到数据,
// 丢弃即可,否则会把已删记录重新写活。
return;
}
boolean sameEpisode = row.getEpisodeId() != null boolean sameEpisode = row.getEpisodeId() != null
&& row.getEpisodeId().equals(state.episodeId()); && row.getEpisodeId().equals(state.episodeId());
if (hidden && sameEpisode) { if (hidden && sameEpisode) {
@@ -661,6 +666,36 @@ public class WatchProgressService {
historyMapper.insert(history); historyMapper.insert(history);
} }
/**
* 把某个身份缓冲里尚未落库的进度立即写进权威表。
* 登录合并前必须先调用:合并只扫库表,缓冲里的设备进度会被漏掉,
* 随后定时落库又会生成一条登录用户看不到的匿名孤儿行。
*
* @return 本次合并的条目数
*/
public int flushIdentity(InteractionIdentity identity) {
WatchProgressBuffer buffer = bufferProvider.getIfAvailable();
if (buffer == null) {
return 0;
}
Map<Long, WatchProgressBuffer.BufferedProgress> states = buffer.getAll(identity);
if (states.isEmpty()) {
return 0;
}
String key = buffer.identityKey(identity);
List<WatchProgressBuffer.IdentityRef> refs = states.keySet().stream()
.map(dramaId -> new WatchProgressBuffer.IdentityRef(
key, identity.userId(), identity.deviceId(), dramaId))
.toList();
try {
return flushRefs(buffer, refs);
} catch (Exception ex) {
// 预写失败不影响合并本身:dirty 队列里的引用还在,下个周期会再落库
log.warn("观看进度缓冲立即落库失败,跳过预写: {}", ex.getMessage());
return 0;
}
}
/** /**
* 登录时合并设备进度和观看流水。 * 登录时合并设备进度和观看流水。
* *
@@ -672,6 +707,9 @@ public class WatchProgressService {
if (userId == null || !StringUtils.hasText(deviceId)) { if (userId == null || !StringUtils.hasText(deviceId)) {
return new MergeResult(0, 0, 0, 0); return new MergeResult(0, 0, 0, 0);
} }
InteractionIdentity deviceIdentity = new InteractionIdentity(null, deviceId);
// 缓冲里可能还有没落库的设备进度,先落库再合并,否则这段进度会被漏掉
flushIdentity(deviceIdentity);
List<WatchProgress> deviceProgress = progressMapper.selectList( List<WatchProgress> deviceProgress = progressMapper.selectList(
Wrappers.<WatchProgress>lambdaQuery() Wrappers.<WatchProgress>lambdaQuery()
.isNull(WatchProgress::getUserId) .isNull(WatchProgress::getUserId)
@@ -731,6 +769,11 @@ public class WatchProgressService {
} }
historyMapper.deleteById(device.getId()); historyMapper.deleteById(device.getId());
} }
// 设备侧进度已迁走或淘汰,缓冲留着只会让读覆盖层把已删行读回来
WatchProgressBuffer buffer = bufferProvider.getIfAvailable();
if (buffer != null) {
buffer.clear(deviceIdentity);
}
return new MergeResult( return new MergeResult(
progressMigrated, progressIgnored, historyMigrated, historyIgnored); progressMigrated, progressIgnored, historyMigrated, historyIgnored);
} }