From a71d2701d1cd2e38c2f9fbab0cf9096a12e273d4 Mon Sep 17 00:00:00 2001 From: Zhou Weidong <502333560@qq.com> Date: Wed, 28 Oct 2026 19:02:32 +0800 Subject: [PATCH] =?UTF-8?q?fix(playback):=20=E7=99=BB=E5=BD=95=E5=90=88?= =?UTF-8?q?=E5=B9=B6=E5=89=8D=E9=A2=84=E5=86=99=E8=AE=BE=E5=A4=87=E7=BC=93?= =?UTF-8?q?=E5=86=B2=EF=BC=8C=E8=BF=9F=E5=88=B0=E6=9D=A1=E7=9B=AE=E4=B8=8D?= =?UTF-8?q?=E5=BE=97=E5=A4=8D=E6=B4=BB=E5=B7=B2=E5=88=A0=E8=AE=B0=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 flushIdentity：把某身份缓冲里未落库的进度立即写进权威表； mergeDeviceIntoUser 先调用它再扫库，否则登录前 2 秒内的上报会被漏掉， 稍后落库还会生成一条登录用户看不到的匿名孤儿行；合并完清掉设备侧缓冲。 - planOne 增加判定：逻辑删除会刷新 updated_at，上报时间不晚于它的条目直接丢弃， 防止删除瞬间已 drain 的在途引用把 deleted 写回 0。 - 去掉 save(Long userId, ...) 上多余的 @Transactional（缓冲路径白开事务）。 --- .../service/WatchProgressService.java | 45 ++++++++++++++++++- 1 file changed, 44 insertions(+), 1 deletion(-) diff --git a/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressService.java b/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressService.java index cd4246b..607ffac 100644 --- a/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressService.java +++ b/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressService.java @@ -85,7 +85,6 @@ public class WatchProgressService { .expireAfterWrite(Duration.ofMinutes(5)) .build(); - @Transactional(rollbackFor = Exception.class) public ProgressRespVO save(Long userId, ProgressSaveReqVO req) { return save(new InteractionIdentity(userId, ""), req); } @@ -460,6 +459,12 @@ public class WatchProgressService { return; } 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 && row.getEpisodeId().equals(state.episodeId()); if (hidden && sameEpisode) { @@ -661,6 +666,36 @@ public class WatchProgressService { historyMapper.insert(history); } + /** + * 把某个身份缓冲里尚未落库的进度立即写进权威表。 + * 登录合并前必须先调用：合并只扫库表，缓冲里的设备进度会被漏掉， + * 随后定时落库又会生成一条登录用户看不到的匿名孤儿行。 + * + * @return 本次合并的条目数 + */ + public int flushIdentity(InteractionIdentity identity) { + WatchProgressBuffer buffer = bufferProvider.getIfAvailable(); + if (buffer == null) { + return 0; + } + Map states = buffer.getAll(identity); + if (states.isEmpty()) { + return 0; + } + String key = buffer.identityKey(identity); + List 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)) { return new MergeResult(0, 0, 0, 0); } + InteractionIdentity deviceIdentity = new InteractionIdentity(null, deviceId); + // 缓冲里可能还有没落库的设备进度，先落库再合并，否则这段进度会被漏掉 + flushIdentity(deviceIdentity); List deviceProgress = progressMapper.selectList( Wrappers.lambdaQuery() .isNull(WatchProgress::getUserId) @@ -731,6 +769,11 @@ public class WatchProgressService { } historyMapper.deleteById(device.getId()); } + // 设备侧进度已迁走或淘汰，缓冲留着只会让读覆盖层把已删行读回来 + WatchProgressBuffer buffer = bufferProvider.getIfAvailable(); + if (buffer != null) { + buffer.clear(deviceIdentity); + } return new MergeResult( progressMigrated, progressIgnored, historyMigrated, historyIgnored); }