update: implement SABR recovery mechanism for stream stalls

Author: 1Lucas1apk

src/sources/youtube/YouTube.js

@@ -930,6 +930,8 @@ export default class YouTubeSource {
         })
 
         const stream = new PassThrough()
+        let isRecovering = false
+        let lastRecoverAt = 0
 
         sabr.on('data', (chunk) => {
           if (!stream.write(chunk)) {
@@ -940,6 +942,48 @@ export default class YouTubeSource {
 
         sabr.on('end', () => stream.end())
         sabr.on('finishBuffering', () => stream.emit('finishBuffering'))
+        sabr.on('stall', async () => {
+          if (isRecovering || stream.destroyed) return
+
+          const now = Date.now()
+          if (now - lastRecoverAt < 2000) return
+          lastRecoverAt = now
+
+          isRecovering = true
+          try {
+            logger(
+              'warn',
+              'YouTube',
+              `SABR stall detected for ${decodedTrack.title}. Refreshing session...`
+            )
+            const newUrlData = await this.getTrackUrl(decodedTrack, null, true)
+            if (!newUrlData || newUrlData.protocol !== 'sabr') {
+              throw new Error('No SABR session available for recovery')
+            }
+
+            const ad = newUrlData.additionalData || {}
+            sabr.clearBuffers()
+            sabr.updateSession({
+              serverAbrStreamingUrl: ad.serverAbrStreamingUrl || newUrlData.url,
+              videoPlaybackUstreamerConfig: ad.videoPlaybackUstreamerConfig,
+              poToken: ad.poToken,
+              visitorData: ad.visitorData,
+              clientInfo: ad.clientInfo,
+              formats: ad.formats,
+              userAgent: ad.userAgent,
+              playbackCookie: ad.playbackCookie
+            })
+          } catch (err) {
+            logger(
+              'warn',
+              'YouTube',
+              `SABR recovery failed: ${err.message}`
+            )
+            if (!stream.destroyed) stream.destroy(err)
+          } finally {
+            isRecovering = false
+          }
+        })
         sabr.on('error', async (err) => {
           logger('error', 'YouTube', `SABR stream error: ${err.message}`)
 

src/sources/youtube/sabr/sabr.js

@@ -282,6 +282,7 @@ export class SabrStream extends PassThrough {
         this.startTime = config.startTime || 0;
         this.positionCallback = config.positionCallback;
         this.userAgent = config.userAgent || USER_AGENT;
+        this.recoveryPending = false;
 
 
 
@@ -354,6 +355,10 @@ export class SabrStream extends PassThrough {
         try {
             if (this.lastVirtualAdvanceAt === 0) this.lastVirtualAdvanceAt = Date.now();
             while (!this._aborted && !this.destroyed && !this.streamFinished) {
+                if (this.recoveryPending) {
+                    await wait(500, signal);
+                    continue;
+                }
                 if (this.requestNumber === 0) {
                     try {
                         const tokenData = await poTokenManager.generate(this.videoId, this.visitorData);
@@ -485,6 +490,18 @@ export class SabrStream extends PassThrough {
                 logger('error', 'SABR', `Failed to decode PO token (session update): ${e.message}`);
             }
         }
+        if (config.visitorData) {
+            this.visitorData = config.visitorData;
+        }
+        if (config.clientInfo) {
+            this.clientInfo = config.clientInfo;
+        }
+        if (config.formats) {
+            this.formatIds = config.formats;
+        }
+        if (config.userAgent) {
+            this.userAgent = config.userAgent;
+        }
         if (config.playbackCookie) {
             if (!this.nextRequestPolicy) this.nextRequestPolicy = {};
             this.nextRequestPolicy.playbackCookie = config.playbackCookie;
@@ -494,6 +511,7 @@ export class SabrStream extends PassThrough {
         this.requestNumber = 0;
         this.noMediaStreak = 0;
         this.pendingRangesHeaders.clear();
+        this.recoveryPending = false;
 
         logger('info', 'SABR', `Session updated. Continuing with RN=${this.requestNumber}, URL=${this.serverAbrStreamingUrl.slice(0, 50)}...`);
     }
@@ -616,6 +634,7 @@ export class SabrStream extends PassThrough {
         if (status.status === 2) {
             logger('warn', 'SABR', `Stream Protection Status: ${status.status} (Limited Playback). Triggering token refresh...`);
             poTokenManager.reset();
+            this.recoveryPending = true;
             this.emit('stall');
             return;
         }