Fix infinite 401 retry loop when app logs resubscribe fails

packages/app/src/cli/services/app-logs/dev/poll-app-logs.test.ts

@@ -1,6 +1,7 @@
 import {pollAppLogs} from './poll-app-logs.js'
 import {writeAppLogsToFile} from './write-app-logs.js'
 import {FunctionRunLog} from '../types.js'
+import {MAX_CONSECUTIVE_RESUBSCRIBE_FAILURES} from '../utils.js'
 import {testDeveloperPlatformClient} from '../../../models/app/app.test-data.js'
 import {describe, expect, test, vi, beforeEach, afterEach} from 'vitest'
 import * as components from '@shopify/cli-kit/node/ui/components'
@@ -385,6 +386,36 @@ describe('pollAppLogs', () => {
     expect(vi.getTimerCount()).toEqual(1)
   })
 
+  test('retries at 60s interval when resubscribe throws on 401', async () => {
+    // Given
+    const outputDebugSpy = vi.spyOn(output, 'outputDebug')
+    const outputWarnSpy = vi.spyOn(output, 'outputWarn')
+    const timeoutSpy = vi.spyOn(global, 'setTimeout')
+    const response = {errors: ['Unauthorized'], status: 401}
+    const mockedDeveloperPlatformClient = testDeveloperPlatformClient({
+      appLogs: vi.fn().mockResolvedValue(response),
+    })
+    const failingResubscribe = vi.fn().mockRejectedValue(new Error('Network error'))
+
+    // When
+    await pollAppLogs({
+      stdout,
+      appLogsFetchInput: {jwtToken: JWT_TOKEN},
+      developerPlatformClient: mockedDeveloperPlatformClient,
+      resubscribeCallback: failingResubscribe,
+      storeName: 'storeName',
+      organizationId: 'organizationId',
+      logsDir: TEST_LOGS_DIR,
+    })
+
+    // Then
+    expect(failingResubscribe).toHaveBeenCalled()
+    expect(outputDebugSpy).toHaveBeenCalledWith(expect.stringContaining('Failed to resubscribe'))
+    expect(outputWarnSpy).toHaveBeenCalledWith('Request throttled while polling app logs.', stdout)
+    expect(timeoutSpy).toHaveBeenCalledWith(expect.any(Function), 60000)
+    expect(vi.getTimerCount()).toEqual(1)
+  })
+
   test('displays error message, waits, and retries if error occurred', async () => {
     // Given
     const outputDebugSpy = vi.spyOn(output, 'outputDebug')
@@ -412,6 +443,35 @@ describe('pollAppLogs', () => {
     expect(vi.getTimerCount()).toEqual(1)
   })
 
+  test('stops polling after MAX consecutive resubscribe failures', async () => {
+    // Given
+    const outputWarnSpy = vi.spyOn(output, 'outputWarn')
+    const response = {errors: ['Unauthorized'], status: 401}
+    const mockedDeveloperPlatformClient = testDeveloperPlatformClient({
+      appLogs: vi.fn().mockResolvedValue(response),
+    })
+    const failingResubscribe = vi.fn().mockRejectedValue(new Error('Network error'))
+
+    // When - start with failures already at MAX - 1
+    await pollAppLogs({
+      stdout,
+      appLogsFetchInput: {jwtToken: JWT_TOKEN},
+      developerPlatformClient: mockedDeveloperPlatformClient,
+      resubscribeCallback: failingResubscribe,
+      storeName: 'storeName',
+      organizationId: 'organizationId',
+      logsDir: TEST_LOGS_DIR,
+      consecutiveResubscribeFailures: MAX_CONSECUTIVE_RESUBSCRIBE_FAILURES - 1,
+    })
+
+    // Then - should output terminal message and NOT schedule a timer
+    expect(outputWarnSpy).toHaveBeenCalledWith(
+      'App log streaming session has expired. Please restart your dev session.',
+      stdout,
+    )
+    expect(vi.getTimerCount()).toEqual(0)
+  })
+
   test('displays error message, waits, and retries if response contained bad JSON', async () => {
     // Given
     const outputDebugSpy = vi.spyOn(output, 'outputDebug')

packages/app/src/cli/services/app-logs/dev/poll-app-logs.ts

@@ -10,6 +10,7 @@ import {
   LOG_TYPE_REQUEST_EXECUTION,
   REQUEST_EXECUTION_IN_BACKGROUND_NO_CACHED_RESPONSE_REASON,
   REQUEST_EXECUTION_IN_BACKGROUND_CACHE_ABOUT_TO_EXPIRE_REASON,
+  MAX_CONSECUTIVE_RESUBSCRIBE_FAILURES,
   handleFetchAppLogsError,
   AppLogsOptions,
 } from '../utils.js'
@@ -29,6 +30,7 @@ export const pollAppLogs = async ({
   organizationId,
   abortSignal,
   logsDir,
+  consecutiveResubscribeFailures = 0,
 }: {
   stdout: Writable
   appLogsFetchInput: AppLogsOptions
@@ -38,6 +40,7 @@ export const pollAppLogs = async ({
   organizationId: string
   abortSignal?: AbortSignal
   logsDir: string
+  consecutiveResubscribeFailures?: number
 }) => {
   if (abortSignal?.aborted) {
     return
@@ -46,11 +49,13 @@ export const pollAppLogs = async ({
   try {
     let nextJwtToken = jwtToken
     let retryIntervalMs = POLLING_INTERVAL_MS
+    let nextConsecutiveResubscribeFailures = consecutiveResubscribeFailures
 
     const response = await developerPlatformClient.appLogs({jwtToken, cursor}, organizationId)
 
     const {errors, status} = response as AppLogsError
     if (status === 200) {
+      nextConsecutiveResubscribeFailures = 0
       const {app_logs: appLogs} = response as AppLogsSuccess
 
       for (const log of appLogs) {
@@ -102,6 +107,16 @@ export const pollAppLogs = async ({
         },
       })
 
+      if (result.resubscribeFailed) {
+        nextConsecutiveResubscribeFailures += 1
+        if (nextConsecutiveResubscribeFailures >= MAX_CONSECUTIVE_RESUBSCRIBE_FAILURES) {
+          outputWarn('App log streaming session has expired. Please restart your dev session.', stdout)
+          return
+        }
+      } else {
+        nextConsecutiveResubscribeFailures = 0
+      }
+
       if (result.nextJwtToken) {
         nextJwtToken = result.nextJwtToken
       }
@@ -123,6 +138,7 @@ export const pollAppLogs = async ({
         organizationId,
         abortSignal,
         logsDir,
+        consecutiveResubscribeFailures: nextConsecutiveResubscribeFailures,
       }).catch((error) => {
         outputDebug(`Unexpected error during polling: ${error}}\n`)
       })

packages/app/src/cli/services/app-logs/logs-command/render-json-logs.test.ts

@@ -1,6 +1,6 @@
 import {renderJsonLogs} from './render-json-logs.js'
 import {pollAppLogs} from './poll-app-logs.js'
-import {handleFetchAppLogsError} from '../utils.js'
+import {handleFetchAppLogsError, MAX_CONSECUTIVE_RESUBSCRIBE_FAILURES} from '../utils.js'
 import {testDeveloperPlatformClient} from '../../../models/app/app.test-data.js'
 import {outputInfo, outputResult} from '@shopify/cli-kit/node/output'
 import {describe, expect, vi, test, beforeEach, afterEach} from 'vitest'
@@ -101,6 +101,67 @@
     expect(outputResult).not.toHaveBeenCalled()
   })
 
+  test('should retry at throttle interval when handleFetchAppLogsError returns null token', async () => {
+    const mockErrorResponse = {
+      errors: [{status: 401, message: 'Unauthorized'}],
+    }
+    const pollAppLogsMock = vi.fn().mockResolvedValue(mockErrorResponse)
+    vi.mocked(pollAppLogs).mockImplementation(pollAppLogsMock)
+    const throttleRetryInterval = 60000
+    const handleFetchAppLogsErrorMock = vi.fn(() => {
+      return Promise.resolve({nextJwtToken: null, retryIntervalMs: throttleRetryInterval, resubscribeFailed: false})
+    })
+    vi.mocked(handleFetchAppLogsError).mockImplementation(handleFetchAppLogsErrorMock)
+
+    const storeNameById = new Map<string, string>()
+    storeNameById.set('1', 'storeName')
+    await renderJsonLogs({
+      pollOptions: {cursor: 'cursor', filters: {status: undefined, sources: undefined}, jwtToken: 'jwtToken'},
+      options: {
+        variables: {shopIds: [], apiKey: ''},
+        developerPlatformClient: testDeveloperPlatformClient(),
+      },
+      storeNameById,
+      organizationId: 'organizationId',
+    })
+
+    expect(handleFetchAppLogsError).toHaveBeenCalled()
+    expect(pollAppLogs).toHaveBeenCalled()
+    expect(setTimeout).toHaveBeenCalledWith(expect.any(Function), throttleRetryInterval)
+    expect(vi.getTimerCount()).toEqual(1)
+  })
+
+  test('should stop polling after MAX consecutive resubscribe failures', async () => {
+    const mockErrorResponse = {
+      errors: [{status: 401, message: 'Unauthorized'}],
+    }
+    const pollAppLogsMock = vi.fn().mockResolvedValue(mockErrorResponse)
+    vi.mocked(pollAppLogs).mockImplementation(pollAppLogsMock)
+    const handleFetchAppLogsErrorMock = vi.fn(() => {
+      return Promise.resolve({nextJwtToken: null, retryIntervalMs: 60000, resubscribeFailed: true})
+    })
+    vi.mocked(handleFetchAppLogsError).mockImplementation(handleFetchAppLogsErrorMock)
+
+    const storeNameById = new Map<string, string>()
+    storeNameById.set('1', 'storeName')
+    await renderJsonLogs({
+      pollOptions: {cursor: 'cursor', filters: {status: undefined, sources: undefined}, jwtToken: 'jwtToken'},
+      options: {
+        variables: {shopIds: [], apiKey: ''},
+        developerPlatformClient: testDeveloperPlatformClient(),
+      },
+      storeNameById,
+      organizationId: 'organizationId',
+      consecutiveResubscribeFailures: MAX_CONSECUTIVE_RESUBSCRIBE_FAILURES - 1,
+    })
+
+    expect(handleFetchAppLogsError).toHaveBeenCalled()
+    expect(outputInfo).toHaveBeenCalledWith(
+      JSON.stringify({message: 'App log streaming session has expired. Please restart your dev session.'}),
+    )
+    expect(vi.getTimerCount()).toEqual(0)
+  })
+
   test('should handle error response and retry as expected', async () => {
     const mockErrorResponse = {
       errors: [{status: 500, message: 'Server Error'}],
@@ -110,9 +171,11 @@
     const mockRetryInterval = 1000
     const handleFetchAppLogsErrorMock = vi.fn((input) => {
       input.onUnknownError(mockRetryInterval)
-      return new Promise<{retryIntervalMs: number; nextJwtToken: string | null}>((resolve, _reject) => {
-        resolve({nextJwtToken: 'new-jwt-token', retryIntervalMs: mockRetryInterval})
-      })
+      return new Promise<{retryIntervalMs: number; nextJwtToken: string | null; resubscribeFailed: boolean}>(
+        (resolve, _reject) => {
+          resolve({nextJwtToken: 'new-jwt-token', retryIntervalMs: mockRetryInterval, resubscribeFailed: false})
+        },
+      )
     })
     vi.mocked(handleFetchAppLogsError).mockImplementation(handleFetchAppLogsErrorMock)
 

packages/app/src/cli/services/app-logs/logs-command/render-json-logs.ts

@@ -2,6 +2,7 @@ import {pollAppLogs} from './poll-app-logs.js'
 import {PollOptions, SubscribeOptions, ErrorResponse, SuccessResponse} from '../types.js'
 import {
   POLLING_INTERVAL_MS,
+  MAX_CONSECUTIVE_RESUBSCRIBE_FAILURES,
   handleFetchAppLogsError,
   subscribeToAppLogs,
   toFormattedAppLogJson,
@@ -14,15 +15,18 @@ export async function renderJsonLogs({
   options: {variables, developerPlatformClient},
   storeNameById,
   organizationId,
+  consecutiveResubscribeFailures = 0,
 }: {
   pollOptions: PollOptions
   options: SubscribeOptions
   storeNameById: Map<string, string>
   organizationId: string
+  consecutiveResubscribeFailures?: number
 }): Promise<void> {
   const response = await pollAppLogs({pollOptions, developerPlatformClient, organizationId})
   let retryIntervalMs = POLLING_INTERVAL_MS
   let nextJwtToken = pollOptions.jwtToken
+  let nextConsecutiveResubscribeFailures = consecutiveResubscribeFailures
 
   const errorResponse = response as ErrorResponse
 
@@ -40,10 +44,22 @@ export async function renderJsonLogs({
       },
     })
 
+    if (result.resubscribeFailed) {
+      nextConsecutiveResubscribeFailures += 1
+      if (nextConsecutiveResubscribeFailures >= MAX_CONSECUTIVE_RESUBSCRIBE_FAILURES) {
+        outputInfo(JSON.stringify({message: 'App log streaming session has expired. Please restart your dev session.'}))
+        return
+      }
+    } else {
+      nextConsecutiveResubscribeFailures = 0
+    }
+
     if (result.nextJwtToken) {
       nextJwtToken = result.nextJwtToken
     }
     retryIntervalMs = result.retryIntervalMs
+  } else {
+    nextConsecutiveResubscribeFailures = 0
   }
 
   const {cursor: nextCursor, appLogs} = response as SuccessResponse
@@ -76,6 +92,7 @@ export async function renderJsonLogs({
       },
       storeNameById,
       organizationId,
+      consecutiveResubscribeFailures: nextConsecutiveResubscribeFailures,
     }).catch((error) => {
       throw error
     })

packages/app/src/cli/services/app-logs/logs-command/ui/components/hooks/usePollAppLogs.test.tsx

@@ -7,6 +7,7 @@ import {
   POLLING_ERROR_RETRY_INTERVAL_MS,
   POLLING_INTERVAL_MS,
   POLLING_THROTTLE_RETRY_INTERVAL_MS,
+  MAX_CONSECUTIVE_RESUBSCRIBE_FAILURES,
   parseFunctionRunPayload,
 } from '../../../../utils.js'
 import {
@@ -502,6 +503,46 @@ describe('usePollAppLogs', () => {
     expect(hook.lastResult?.errors).toHaveLength(0)
   })
 
+  test('stops polling after MAX consecutive resubscribe failures', async () => {
+    const mockedPollAppLogs = vi.fn().mockResolvedValue(POLL_APP_LOGS_FOR_LOGS_401_RESPONSE)
+    vi.mocked(pollAppLogs).mockImplementation(mockedPollAppLogs)
+
+    const mockedDeveloperPlatformClient = testDeveloperPlatformClient()
+    const resubscribeCallback = vi.fn().mockRejectedValue(new Error('Session expired'))
+
+    const hook = renderHook(() =>
+      usePollAppLogs({
+        initialJwt: MOCKED_JWT_TOKEN,
+        filters: EMPTY_FILTERS,
+        resubscribeCallback,
+        storeNameById: STORE_NAME_BY_ID,
+        developerPlatformClient: mockedDeveloperPlatformClient,
+        organizationId: MOCKED_ORGANIZATION_ID,
+      }),
+    )
+
+    // needed to await the render
+    await vi.advanceTimersByTimeAsync(0)
+
+    // Wait for the first poll
+    await waitForMockCalls(mockedPollAppLogs, 1)
+
+    // Advance through MAX_CONSECUTIVE_RESUBSCRIBE_FAILURES - 1 more polls
+    for (let i = 1; i < MAX_CONSECUTIVE_RESUBSCRIBE_FAILURES; i++) {
+      // eslint-disable-next-line no-await-in-loop
+      await vi.advanceTimersToNextTimerAsync()
+      // eslint-disable-next-line no-await-in-loop
+      await waitForMockCalls(mockedPollAppLogs, i + 1)
+    }
+
+    // Flush React 19 batched state updates
+    await vi.advanceTimersByTimeAsync(0)
+
+    expect(hook.lastResult?.errors).toEqual(['App log streaming session has expired. Please restart your dev session.'])
+    // Polling should have stopped - no more timers scheduled
+    expect(vi.getTimerCount()).toEqual(0)
+  })
+
   test("ignores logs from stores that don't have a matching shop name", async () => {
     const mockedPollAppLogs = vi.fn().mockResolvedValue(POLL_APP_LOGS_FOR_LOGS_RESPONSE)
     vi.mocked(pollAppLogs).mockImplementation(mockedPollAppLogs)