@@ -23,6 +23,7 @@ const TRANSPORT_DIAGNOSTICS_POLL_INTERVAL_MS = 2000;
2323const EMPTY_BATCH_BACKFILL_COOLDOWN_MS = 1000 ;
2424const EMPTY_BATCH_BACKFILL_TRIGGER = 4 ;
2525const BACKFILL_STALE_THRESHOLD_NS = 1_000_000_000n ;
26+ const BACKFILL_STALE_TOPIC_PERIOD_MULTIPLIER = 3 ;
2627const STALE_TOPIC_REFRESH_COOLDOWN_MS = 500 ;
2728const SLOW_DISTRIBUTION_MS = 16 ;
2829
@@ -162,6 +163,7 @@ export class IterablePlayer implements Player {
162163 private _fallbackBackfillCount = 0 ;
163164 private _lastFallbackBackfillMs = 0 ;
164165 private _lastStaleRefreshMs = 0 ;
166+ private _staleRefreshInFlight = false ;
165167 private _isBuffering = false ;
166168 private _topicLastMessageNs = new Map < string , bigint > ( ) ;
167169 private _highFrequencyConsumerSignature = "" ;
@@ -1004,10 +1006,7 @@ export class IterablePlayer implements Player {
10041006 }
10051007 this . _currentTime = nextTime ;
10061008 this . _clock . seek ( this . _currentTime , performance . now ( ) ) ;
1007- await this . _refreshStaleTopicsFromBackfill ( now , epoch ) ;
1008- if ( ! this . _isPlaybackEpochCurrent ( epoch ) ) {
1009- return ;
1010- }
1009+ this . _scheduleStaleTopicsRefresh ( now , epoch ) ;
10111010
10121011 if ( this . _initialization && toNano ( this . _currentTime ) >= toNano ( this . _initialization . end ) ) {
10131012 if ( this . _isLooping ) {
@@ -1238,7 +1237,10 @@ export class IterablePlayer implements Player {
12381237 this . _cursor = cursor ;
12391238 }
12401239
1241- private async _refreshStaleTopicsFromBackfill ( nowMs : number , epoch : number ) : Promise < void > {
1240+ private _scheduleStaleTopicsRefresh ( nowMs : number , epoch : number ) : void {
1241+ if ( this . _staleRefreshInFlight ) {
1242+ return ;
1243+ }
12421244 if ( nowMs - this . _lastStaleRefreshMs < STALE_TOPIC_REFRESH_COOLDOWN_MS ) {
12431245 return ;
12441246 }
@@ -1250,40 +1252,56 @@ export class IterablePlayer implements Player {
12501252 const staleTopics = topics . filter ( ( topic ) => {
12511253 const lastNs = this . _topicLastMessageNs . get ( topic ) ;
12521254 if ( lastNs == null ) return true ;
1253- return nowNs - lastNs > BACKFILL_STALE_THRESHOLD_NS ;
1255+ return nowNs - lastNs > this . _staleThresholdNsForTopic ( topic ) ;
12541256 } ) ;
12551257 if ( staleTopics . length === 0 ) {
12561258 return ;
12571259 }
12581260 this . _lastStaleRefreshMs = nowMs ;
12591261 const referenceTime = this . _currentTime ;
1260- try {
1261- const messages = await this . _source . getBackfillMessages ( {
1262- time : referenceTime ,
1263- topics : staleTopics ,
1264- } ) ;
1265- if ( ! this . _isPlaybackEpochCurrent ( epoch ) ) {
1266- return ;
1267- }
1268- // Same reasoning as in _handleEmptyBatch: latched topics would otherwise
1269- // get re-delivered on every refresh tick (~5 Hz), causing panels like
1270- // the 3D/URDF renderer to rebuild from scratch and leak GPU buffers.
1271- const freshMessages = this . _filterAlreadyDeliveredMessages ( messages ) ;
1272- if ( freshMessages . length === 0 ) {
1273- return ;
1274- }
1275- this . _distributeMessages ( freshMessages , referenceTime ) ;
1276- if ( this . _debugEnabled ) {
1277- console . debug ( "[Playback] stale refresh " + JSON . stringify ( {
1278- staleTopicCount : staleTopics . length ,
1279- messageCount : messages . length ,
1280- freshCount : freshMessages . length ,
1281- currentTime : this . _currentTime ,
1282- } ) ) ;
1262+ this . _staleRefreshInFlight = true ;
1263+ void ( async ( ) => {
1264+ try {
1265+ const messages = await this . _source . getBackfillMessages ( {
1266+ time : referenceTime ,
1267+ topics : staleTopics ,
1268+ } ) ;
1269+ if ( ! this . _isPlaybackEpochCurrent ( epoch ) ) {
1270+ return ;
1271+ }
1272+ // Same reasoning as in _handleEmptyBatch: latched topics would otherwise
1273+ // get re-delivered on every refresh tick (~5 Hz), causing panels like
1274+ // the 3D/URDF renderer to rebuild from scratch and leak GPU buffers.
1275+ const freshMessages = this . _filterAlreadyDeliveredMessages ( messages ) ;
1276+ if ( freshMessages . length === 0 ) {
1277+ return ;
1278+ }
1279+ this . _distributeMessages ( freshMessages , referenceTime ) ;
1280+ if ( this . _debugEnabled ) {
1281+ console . debug ( "[Playback] stale refresh " + JSON . stringify ( {
1282+ staleTopicCount : staleTopics . length ,
1283+ messageCount : messages . length ,
1284+ freshCount : freshMessages . length ,
1285+ currentTime : this . _currentTime ,
1286+ } ) ) ;
1287+ }
1288+ } catch ( err ) {
1289+ console . warn ( "IterablePlayer: stale topic refresh failed" , err ) ;
1290+ } finally {
1291+ this . _staleRefreshInFlight = false ;
12831292 }
1284- } catch ( err ) {
1285- console . warn ( "IterablePlayer: stale topic refresh failed" , err ) ;
1293+ } ) ( ) ;
1294+ }
1295+
1296+ private _staleThresholdNsForTopic ( topic : string ) : bigint {
1297+ const stats = this . _initialization ?. topicStats [ topic ] ;
1298+ const frequency = stats ?. frequency ;
1299+ if ( typeof frequency !== "number" || ! Number . isFinite ( frequency ) || frequency <= 0 ) {
1300+ return BACKFILL_STALE_THRESHOLD_NS ;
12861301 }
1302+ const topicPeriodNs = BigInt ( Math . ceil ( 1_000_000_000 / frequency ) ) ;
1303+ const topicThresholdNs = topicPeriodNs * BigInt ( BACKFILL_STALE_TOPIC_PERIOD_MULTIPLIER ) ;
1304+ return topicThresholdNs > BACKFILL_STALE_THRESHOLD_NS ? topicThresholdNs : BACKFILL_STALE_THRESHOLD_NS ;
12871305 }
12881306
12891307 /**
0 commit comments