stop trying to resubscribe after max attempts

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'
@@ -442,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'
@@ -109,7 +109,7 @@
     vi.mocked(pollAppLogs).mockImplementation(pollAppLogsMock)
     const throttleRetryInterval = 60000
     const handleFetchAppLogsErrorMock = vi.fn(() => {
-      return Promise.resolve({nextJwtToken: null, retryIntervalMs: throttleRetryInterval})
+      return Promise.resolve({nextJwtToken: null, retryIntervalMs: throttleRetryInterval, resubscribeFailed: false})
     })
     vi.mocked(handleFetchAppLogsError).mockImplementation(handleFetchAppLogsErrorMock)
 
@@ -127,10 +127,41 @@
 
     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'}],

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,24 @@ 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 +94,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,48 @@ 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)

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

@@ -2,6 +2,7 @@ import {useSelfAdjustingInterval} from './useSelfAdjustingInterval.js'
 import {
   ONE_MILLION,
   POLLING_INTERVAL_MS,
+  MAX_CONSECUTIVE_RESUBSCRIBE_FAILURES,
   parseFunctionRunPayload,
   LOG_TYPE_FUNCTION_RUN,
   LOG_TYPE_RESPONSE_FROM_CACHE,
@@ -57,6 +58,7 @@ async function performPoll({
     organizationId,
   })
 
+  let resubscribeFailed = false
   const errorResponse = response as ErrorResponse
 
   if (errorResponse.errors) {
@@ -73,6 +75,8 @@ async function performPoll({
       },
     })
 
+    resubscribeFailed = result.resubscribeFailed
+
     if (result.nextJwtToken) {
       nextJwtToken = result.nextJwtToken
     }
@@ -134,7 +138,7 @@ async function performPoll({
     }
   }
 
-  return {nextJwtToken, retryIntervalMs, cursor: nextCursor ?? cursor}
+  return {nextJwtToken, retryIntervalMs, cursor: nextCursor ?? cursor, resubscribeFailed}
 }
 
 export function usePollAppLogs({
@@ -150,6 +154,7 @@ export function usePollAppLogs({
   const nextJwtToken = useRef(initialJwt)
   const retryIntervalMs = useRef(0)
   const cursor = useRef<string | undefined>('')
+  const consecutiveResubscribeFailures = useRef(0)
 
   const performPollCallback = useCallback(async () => {
     const res = await performPoll({
@@ -164,6 +169,16 @@ export function usePollAppLogs({
       organizationId,
     })
 
+    if (res.resubscribeFailed) {
+      consecutiveResubscribeFailures.current += 1
+      if (consecutiveResubscribeFailures.current >= MAX_CONSECUTIVE_RESUBSCRIBE_FAILURES) {
+        setErrors(['App log streaming session has expired. Please restart your dev session.'])
+        return {retryIntervalMs: 0}
+      }
+    } else {
+      consecutiveResubscribeFailures.current = 0
+    }
+
     // ESLint is concerned about these updates being atomic, but the approach to useSelfAdjustingInterval ensures that is the case.
     // eslint-disable-next-line require-atomic-updates
     nextJwtToken.current = res.nextJwtToken

packages/app/src/cli/services/app-logs/utils.ts

@@ -20,6 +20,7 @@ import {Writable} from 'stream'
 export const POLLING_INTERVAL_MS = 450
 export const POLLING_ERROR_RETRY_INTERVAL_MS = 5 * 1000
 export const POLLING_THROTTLE_RETRY_INTERVAL_MS = 60 * 1000
+export const MAX_CONSECUTIVE_RESUBSCRIBE_FAILURES = 5
 export const ONE_MILLION = 1000000
 export const LOG_TYPE_FUNCTION_RUN = 'function_run'
 export const LOG_TYPE_FUNCTION_NETWORK_ACCESS = 'function_network_access'
@@ -109,11 +110,12 @@ export interface AppLogsOptions {
 
 export const handleFetchAppLogsError = async (
   input: FetchAppLogsErrorOptions,
-): Promise<{retryIntervalMs: number; nextJwtToken: string | null}> => {
+): Promise<{retryIntervalMs: number; nextJwtToken: string | null; resubscribeFailed: boolean}> => {
   const {errors} = input.response
 
   let retryIntervalMs = POLLING_INTERVAL_MS
   let nextJwtToken = null
+  let resubscribeFailed = false
 
   if (errors.length > 0) {
     outputDebug(`Errors: ${errors.map((error) => error.message).join(', ')}`)
@@ -124,6 +126,7 @@ export const handleFetchAppLogsError = async (
       } catch (resubscribeError) {
         outputDebug(`Failed to resubscribe to app logs: ${resubscribeError}`)
         retryIntervalMs = POLLING_THROTTLE_RETRY_INTERVAL_MS
+        resubscribeFailed = true
         input.onThrottle(retryIntervalMs)
       }
     } else if (errors.some((error) => error.status === 429)) {
@@ -135,7 +138,7 @@ export const handleFetchAppLogsError = async (
     }
   }
 
-  return {retryIntervalMs, nextJwtToken}
+  return {retryIntervalMs, nextJwtToken, resubscribeFailed}
 }
 
 export function sourcesForApp(app: AppInterface): string[] {