fix(content): fail closed on image inference watchdog timeout, dedupe in-flight inference

docs/MEDIA_PROCESSING.md

@@ -554,6 +554,24 @@ t=8: Debounce fires, process(img) with src=C
 t=9: Inference sent for C, prediction applied ✓
 ```
 
+### Resolving the Source Under Firefox's Pending Request
+
+Every "what is this image's URL" read goes through `resolveImageSource(img)` instead of a bare
+`img.currentSrc || img.src`. Firefox keeps `currentSrc` on the **previous** request until the new
+image's size is known, so right after `img.src = next` the mutation callback still reads the old
+URL. On Google Images that made the placeholder→thumbnail swap (`…&s=10` → `…&s`) look like a no-op
+re-stamp: the 10 px placeholder's `safe` verdict stayed on the element and the full thumbnail
+rendered bare. For images without `srcset`/`<picture>` the reflected `src` is authoritative;
+candidate selection keeps using `currentSrc`. As a second net, `process()` registers one `load`
+listener per image that re-runs `handleSrcChange` whenever the resolved source no longer matches the
+one last processed (also covers srcset re-selection on images that carry no overlay).
+
+Once such a placeholder has decoded, `isPlaceholderResolution` (both natural sides under 32 px, or
+under the host's configured minimum size when that is smaller) finalizes it as `skipped` instead of
+inferring it: a 10 px image carries nothing to detect, and on a single-lane mobile queue those
+placeholders were the bulk of the backlog ahead of real thumbnails. The swap to the real source
+re-enters processing as described above.
+
 ### Robust Image Load Detection
 
 For images that aren't yet loaded, we use **both** `decode()` and `load` event - whichever fires
@@ -682,11 +700,22 @@ owner is stalled (not loaded, e.g. a lazy copy in a hidden subtree) while the ne
 pixels. A duplicate send from a superseded owner is harmless — predictions are keyed by src and the
 second result is idempotent. The entry is deleted when a prediction for that src arrives.
 
-Once a request is actually sent, a 20-second watchdog prevents that entry from living forever if the
-background worker loses the task or inference fails without a prediction broadcast. It retries once
-using a loaded, visible same-source copy when possible; a second timeout finalizes all pending
-copies as `skipped`, matching the pipeline's existing inference-impossible behavior rather than
-leaving Reddit images permanently under the initial blur.
+Once a request is sent, two watchdogs guard the entry. A 120-second send-time guard covers a task
+the background lost before it ever ran. The real 20-second inference watchdog starts only when the
+background broadcasts `status: 'started'` for the src (the task left the queue), because queue wait
+is unbounded on slow devices: single-lane WASM inference on Firefox mobile with a page of offscreen
+Google Images ahead easily exceeded 20 s, and a queue-based timeout revealed the whole below-fold
+band unmasked. Either watchdog retries once using a loaded, visible same-source copy when possible.
+On the second timeout the images **fail closed**: the initial blur stays and the pending entry is
+cleared, so the late prediction still applies when it arrives (`findImagesBySrc` matches blurred
+images) and any later `process()` pass re-sends. Only an explicit `status: 'error'` result from the
+background (inference genuinely impossible) finalizes the copies as `skipped` after two attempts.
+
+The background dedupes as well: a request for a `hostname + src` that is already queued or running
+joins the in-flight task instead of enqueueing a second one (retries, extra `<img>` copies, and
+other tabs on the same host all receive the single broadcast verdict). A joining request with a
+higher priority (a visible copy behind an offscreen one) raises the queued task's priority, and one
+that joins a task already running receives the `started` broadcast again so its watchdog arms.
 
 ### DOM Processing
 

docs/TESTING.md

@@ -218,6 +218,13 @@ Android config expects that action-managed emulator to already be booted when `C
 workflow sets `ANDROID_SINGLE_SESSION=true` so all features run in one WebDriver session, since
 repeated geckodriver sessions can leave Fenix profile resources locked on CI.
 
+Once the session is up, the `before` hook broadcasts a local Nimbus rollout to Fenix's
+`QANimbusToolingReceiver` that turns the `homepage-as-new-tab` feature off. Fenix Nightly enables it
+by default (Bug 1977695), and with it on Fenix opens a fresh homepage tab while Gecko is already
+quitting, which trips `MOZ_RELEASE_ASSERT(domWindow)` in `GeckoViewSupport::Open` and crashes the
+app during geckodriver's session delete. The override has to be sent after launch because
+geckodriver runs `pm clear` on the package while creating the session, wiping any pre-seeded prefs.
+
 **Setup:**
 
 ```bash

entrypoints/background/services/__tests__/inferenceOrchestrationService.test.ts

@@ -0,0 +1,132 @@
+import { describe, expect, it, vi } from 'vitest';
+
+import { InferenceOrchestrationService } from '@/entrypoints/background/services/inferenceOrchestrationService';
+import { QueueService } from '@/entrypoints/background/services/queueService';
+import { DEFAULT_HOST_SETTINGS } from '@/utils/constants/hostsettings';
+
+import type { ImageCacheService } from '@/entrypoints/background/services/imageCacheService';
+import type { IImagePrediction, IMediaMetadata, ImageInferenceResult, InferenceTask } from '@/utils/types';
+
+const { processInferenceTask, gates } = vi.hoisted(() => {
+  const gates = new Map<string, () => void>();
+  const processInferenceTask = vi.fn(
+    (task: { imageSrc: string }) =>
+      new Promise<IImagePrediction>(resolve => {
+        gates.set(task.imageSrc, () =>
+          resolve({
+            src: task.imageSrc,
+            hostname: 'example.com',
+            width: 100,
+            height: 100,
+            predictions: [],
+            timestamp: 0,
+            cacheMetadata: { createdAt: 0, accessedAt: 0 },
+            maskTransform: { scaleX: 1, scaleY: 1, offsetX: 0, offsetY: 0 },
+            processingTime: {
+              fetchTime: 0,
+              decodeTime: 0,
+              queueTime: 0,
+              inferenceTime: 0,
+              e2eTime: 0,
+              backend: 'wasm',
+            },
+            forcedVisibility: 'auto',
+          }),
+        );
+      }),
+  );
+  return { processInferenceTask, gates };
+});
+
+vi.mock('@inference-runtime', () => ({ getCurrentModelId: () => 'test-model' }));
+vi.mock('@/utils/inference', () => ({
+  getBatchCap: () => 1,
+  getInferenceBackend: () => 'wasm',
+  processInferenceBatch: vi.fn(),
+  processInferenceTask,
+}));
+vi.mock('@/entrypoints/background/services/batchCollector', () => ({
+  BatchCollector: class {
+    submit = vi.fn();
+  },
+}));
+
+const cacheService = {
+  getCachedPredictionsBySrc: vi.fn(() => Promise.resolve([])),
+  cachePredictions: vi.fn(() => Promise.resolve()),
+} as unknown as ImageCacheService;
+
+function setup() {
+  const queue = new QueueService();
+  const service = new InferenceOrchestrationService(queue, cacheService);
+  const results: ImageInferenceResult[] = [];
+  service.setOnImagePredictionsCallback(batch => results.push(...batch));
+  const schedule = (imageSrc: string, priority = 0) =>
+    service.scheduleInferenceTask({
+      input: { kind: 'src', imageSrc },
+      hostname: 'example.com',
+      hostSettings: DEFAULT_HOST_SETTINGS,
+      mediaMetadata: { kind: 'image' } as IMediaMetadata,
+      priority,
+    });
+  return { service, queue, results, schedule };
+}
+
+const statuses = (results: ImageInferenceResult[], src: string) =>
+  results
+    .filter(result => (result.status === 'ok' ? result.prediction.src : result.src) === src)
+    .map(result => result.status);
+
+const release = (src: string) => {
+  gates.get(src)?.();
+  gates.delete(src);
+};
+
+describe('InferenceOrchestrationService image dedupe', () => {
+  it('broadcasts started when the task leaves the queue and ok when it completes', async () => {
+    const { results, schedule } = setup();
+    await schedule('a');
+    await vi.waitFor(() => expect(statuses(results, 'a')).toEqual(['started']));
+    release('a');
+    await vi.waitFor(() => expect(statuses(results, 'a')).toEqual(['started', 'ok']));
+    expect(processInferenceTask).toHaveBeenCalledTimes(1);
+  });
+
+  it('joins a queued duplicate instead of running it twice and raises its priority', async () => {
+    const { results, schedule } = setup();
+    await schedule('running');
+    await vi.waitFor(() => expect(statuses(results, 'running')).toEqual(['started']));
+    await schedule('queued', 0);
+    await schedule('queued', 30);
+
+    release('running');
+    await vi.waitFor(() => expect(statuses(results, 'queued')).toEqual(['started']));
+    const queuedTask = processInferenceTask.mock.calls.at(-1)?.[0] as InferenceTask;
+    expect(queuedTask.priority).toBe(30);
+    release('queued');
+    await vi.waitFor(() => expect(statuses(results, 'queued')).toEqual(['started', 'ok']));
+    expect(processInferenceTask).toHaveBeenCalledTimes(2);
+  });
+
+  it('re-broadcasts started for a duplicate that joins an already running task', async () => {
+    const { results, schedule } = setup();
+    await schedule('a');
+    await vi.waitFor(() => expect(statuses(results, 'a')).toEqual(['started']));
+    await schedule('a');
+    expect(statuses(results, 'a')).toEqual(['started', 'started']);
+    release('a');
+    await vi.waitFor(() => expect(statuses(results, 'a')).toEqual(['started', 'started', 'ok']));
+    expect(processInferenceTask).toHaveBeenCalledTimes(1);
+  });
+
+  it('runs the same src again once the previous task has finished', async () => {
+    const { results, schedule } = setup();
+    await schedule('a');
+    await vi.waitFor(() => expect(statuses(results, 'a')).toEqual(['started']));
+    release('a');
+    await vi.waitFor(() => expect(statuses(results, 'a')).toEqual(['started', 'ok']));
+    await schedule('a');
+    await vi.waitFor(() => expect(processInferenceTask).toHaveBeenCalledTimes(2));
+    release('a');
+  });
+});

entrypoints/background/services/__tests__/queueService.test.ts

@@ -40,3 +40,30 @@ describe('QueueService', () => {
     expect(started).toEqual(['running', 'latest-frame']);
   });
 });
+
+describe('QueueService.raisePriority', () => {
+  it('moves a queued task ahead of its siblings', async () => {
+    const queue = new QueueService();
+    const gate = deferred();
+    const started: string[] = [];
+    queue.setTaskProcessingHandler(async queued => {
+      started.push(queued.imageSrc);
+      if (queued.imageSrc === 'running') await gate.promise;
+    });
+
+    const running = queue.enqueue(task('running'));
+    await vi.waitFor(() => expect(started).toEqual(['running']));
+    const first = queue.enqueue(task('first'), undefined, 'first');
+    const second = queue.enqueue(task('second'), undefined, 'second');
+    queue.raisePriority('second', 30);
+
+    gate.resolve();
+    await Promise.all([running, first, second]);
+    expect(started).toEqual(['running', 'second', 'first']);
+  });
+
+  it('ignores an id that is no longer queued', () => {
+    const queue = new QueueService();
+    expect(() => queue.raisePriority('missing', 30)).not.toThrow();
+  });
+});

entrypoints/background/services/inferenceOrchestrationService.ts

@@ -1,4 +1,4 @@
-import { SpanStatusCode, type Counter, type Histogram, type Span } from '@opentelemetry/api';
+import { SpanStatusCode, type Context, type Counter, type Histogram, type Span } from '@opentelemetry/api';
 
 import { getCurrentModelId } from '@inference-runtime';
 
@@ -69,6 +69,14 @@ export type ScheduleArgs = {
   traceparent?: string;
 };
 
+type TaskAttributes = ReturnType<typeof taskAttributes>;
+
+type ImageTaskKey = string & { readonly __brand: 'ImageTaskKey' };
+
+function imageTaskKey(hostname: string, imageSrc: string): ImageTaskKey {
+  return `${hostname}|${imageSrc}` as ImageTaskKey;
+}
+
 const log = getLogger('inferenceOrchestrationService');
 const tracer = getTracer('inference');
 
@@ -123,6 +131,7 @@ export class InferenceOrchestrationService {
   private onGifFramePredictionsCallback?: OnGifFramePredictionsCallback;
   /** At most one not-yet-started playback frame is retained per video session. */
   private queuedPlaybackFrames = new Map<string, { task: InferenceTask; controller: AbortController }>();
+  private readonly inFlightImages = new Map<ImageTaskKey, { task: InferenceTask; started: boolean }>();
   private queueWaitSpans = new WeakMap<InferenceTask, Span>();
 
   // Batches concurrent queue tasks into one session.run for dynamic-batch models.
@@ -196,6 +205,8 @@ export class InferenceOrchestrationService {
       }
     }
 
+    if (mediaMetadata.kind === 'image' && this.joinInFlightImage(args, attributes, traceContext)) return;
+
     const queueStartAt = Date.now();
     const baseTask = {
       imageSrc,
@@ -258,7 +269,10 @@ export class InferenceOrchestrationService {
       task,
       tracer.startSpan(SPAN.queueWait, { attributes: { ...attributes, [ATTR.priority]: args.priority } }, traceContext),
     );
-    this.queueService.enqueue(task, controller?.signal).catch(error => {
+    const inFlightKey = mediaMetadata.kind === 'image' ? imageTaskKey(hostname, imageSrc) : undefined;
+    if (inFlightKey) this.inFlightImages.set(inFlightKey, { task, started: false });
+    this.queueService.enqueue(task, controller?.signal, inFlightKey).catch(error => {
+      if (inFlightKey) this.inFlightImages.delete(inFlightKey);
       if (controller?.signal.aborted) {
         this.endQueueWait(task, 'aborted');
         return;
@@ -269,6 +283,26 @@ export class InferenceOrchestrationService {
     });
   }
 
+  private joinInFlightImage(args: ScheduleArgs, attributes: TaskAttributes, traceContext?: Context): boolean {
+    const key = imageTaskKey(args.hostname, args.input.imageSrc);
+    const inFlight = this.inFlightImages.get(key);
+    if (!inFlight) return false;
+    if (args.input.kind === 'bitmap') args.input.bitmap.close();
+    if (args.priority > inFlight.task.priority) {
+      inFlight.task.priority = args.priority;
+      this.queueService.raisePriority(key, args.priority);
+    }
+    if (inFlight.started) this.notifyImageInferenceStarted(inFlight.task);
+    log.debug('inference.image.joined', { ...attributes, [ATTR.priority]: args.priority }, traceContext);
+    return true;
+  }
+
+  private releaseInFlightImage(task: InferenceTask): void {
+    if (task.mediaMetadata.kind !== 'image') return;
+    const key = imageTaskKey(task.hostname, task.imageSrc);
+    if (this.inFlightImages.get(key)?.task === task) this.inFlightImages.delete(key);
+  }
+
   private endQueueWait(task: InferenceTask, outcome: 'started' | 'aborted' | 'superseded' | 'error'): void {
     const span = this.queueWaitSpans.get(task);
     if (!span) return;
@@ -281,6 +315,7 @@ export class InferenceOrchestrationService {
     this.queueService.setTaskProcessingHandler(async (task: InferenceTask) => {
       this.markPlaybackFrameStarted(task);
       this.endQueueWait(task, 'started');
+      this.notifyImageInferenceStarted(task);
       const attributes = taskAttributes(task);
       const runSpan = tracer.startSpan(SPAN.run, { attributes }, task.traceContext);
       const runStartedAt = Date.now();
@@ -305,6 +340,8 @@ export class InferenceOrchestrationService {
         });
         log.error('inference.run.failed', { ...attributes, error }, task.traceContext);
         this.sendErrorToContent(task, error);
+      } finally {
+        this.releaseInFlightImage(task);
       }
     });
   }
@@ -462,6 +499,23 @@ export class InferenceOrchestrationService {
     }
   }
 
+  private notifyImageInferenceStarted(task: InferenceTask): void {
+    if (task.mediaMetadata.kind !== 'image') return;
+    const inFlight = this.inFlightImages.get(imageTaskKey(task.hostname, task.imageSrc));
+    if (inFlight?.task === task) inFlight.started = true;
+    this.sendImageResultsToContent(
+      [
+        {
+          status: 'started',
+          src: task.imageSrc,
+          hostname: task.hostname,
+          traceparent: injectTraceparent(task.traceContext),
+        },
+      ],
+      task.hostname,
+    );
+  }
+
   private sendImageResultsToContent(results: ImageInferenceResult[], hostname: string): void {
     try {
       if (this.onImagePredictionsCallback) {

entrypoints/background/services/queueService.ts

@@ -6,6 +6,7 @@ import { type InferenceTask } from '@/utils/types';
 export class QueueService {
   private queue: PQueue;
   private onTaskProcessing?: (task: InferenceTask) => Promise<void>;
+  private readonly queuedIds = new Set<string>();
 
   constructor() {
     this.queue = new PQueue({
@@ -36,15 +37,26 @@ export class QueueService {
     return () => this.queue.off('idle', callback);
   }
 
-  enqueue(task: InferenceTask, signal?: AbortSignal): Promise<void> {
+  enqueue(task: InferenceTask, signal?: AbortSignal, id?: string): Promise<void> {
+    if (id) this.queuedIds.add(id);
     // p-queue: higher priority number = runs first
-    return this.queue.add(
-      async () => {
-        if (this.onTaskProcessing) {
-          await this.onTaskProcessing(task);
-        }
-      },
-      { priority: task.priority, signal },
-    );
+    return this.queue
+      .add(
+        async () => {
+          if (id) this.queuedIds.delete(id);
+          if (this.onTaskProcessing) {
+            await this.onTaskProcessing(task);
+          }
+        },
+        { priority: task.priority, signal, id },
+      )
+      .finally(() => {
+        if (id) this.queuedIds.delete(id);
+      });
+  }
+
+  raisePriority(id: string, priority: number): void {
+    if (!this.queuedIds.has(id)) return;
+    this.queue.setPriority(id, priority);
   }
 }