diff --git a/firebase-crashlytics/firebase-crashlytics.gradle b/firebase-crashlytics/firebase-crashlytics.gradle
index 5b4d09da8a2..3c450c0d68a 100644
--- a/firebase-crashlytics/firebase-crashlytics.gradle
+++ b/firebase-crashlytics/firebase-crashlytics.gradle
@@ -111,4 +111,6 @@ dependencies {
     androidTestImplementation(libs.androidx.test.junit)
     androidTestImplementation(libs.androidx.test.runner)
     androidTestImplementation(libs.truth)
+    androidTestImplementation(libs.playservices.tasks)
+    androidTestImplementation(project(":integ-testing"))
 }
diff --git a/firebase-crashlytics/src/androidTest/java/com/google/firebase/crashlytics/internal/CrashlyticsWorkerTest.java b/firebase-crashlytics/src/androidTest/java/com/google/firebase/crashlytics/internal/CrashlyticsWorkerTest.java
new file mode 100644
index 00000000000..63c9a9ce6d2
--- /dev/null
+++ b/firebase-crashlytics/src/androidTest/java/com/google/firebase/crashlytics/internal/CrashlyticsWorkerTest.java
@@ -0,0 +1,429 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.firebase.crashlytics.internal;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertThrows;
+
+import com.google.android.gms.tasks.Task;
+import com.google.android.gms.tasks.Tasks;
+import com.google.firebase.concurrent.TestOnlyExecutors;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class CrashlyticsWorkerTest {
+  private CrashlyticsWorker crashlyticsWorker;
+
+  @Before
+  public void setUp() {
+    crashlyticsWorker = new CrashlyticsWorker(TestOnlyExecutors.background());
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    // Drain the worker, just in case any test cases would fail but didn't await.
+    crashlyticsWorker.await();
+  }
+
+  @Test
+  public void executesTasksOnThreadPool() throws Exception {
+    Set<String> threads = new HashSet<>();
+
+    // Find thread names by adding the names we touch to the set.
+    for (int i = 0; i < 100; i++) {
+      crashlyticsWorker.submit(() -> threads.add(Thread.currentThread().getName()));
+    }
+
+    crashlyticsWorker.await();
+
+    // Verify that we touched at lease some of the expected background threads.
+    assertThat(threads)
+        .containsAnyOf(
+            "Firebase Background Thread #0",
+            "Firebase Background Thread #1",
+            "Firebase Background Thread #2",
+            "Firebase Background Thread #3");
+  }
+
+  @Test
+  public void executesTasksInOrder() throws Exception {
+    List<Integer> list = new ArrayList<>();
+
+    // Add sequential numbers to the list to validate tasks execute in order.
+    for (int i = 0; i < 100; i++) {
+      int sequential = i;
+      crashlyticsWorker.submit(() -> list.add(sequential));
+    }
+
+    crashlyticsWorker.await();
+
+    // Verify that the tasks executed in order.
+    assertThat(list).isInOrder();
+  }
+
+  @Test
+  public void executesTasksSequentially() throws Exception {
+    List<Integer> list = new ArrayList<>();
+    AtomicBoolean reentrant = new AtomicBoolean(false);
+
+    for (int i = 0; i < 100; i++) {
+      int sequential = i;
+      crashlyticsWorker.submit(
+          () -> {
+            if (reentrant.get()) {
+              // Return early if two runnables ran at the same time.
+              return;
+            }
+
+            reentrant.set(true);
+            // Sleep a bit to simulate some work.
+            sleep(5);
+            list.add(sequential);
+            reentrant.set(false);
+          });
+    }
+
+    crashlyticsWorker.await();
+
+    // Verify that all the runnable tasks executed, one at a time, and in order.
+    assertThat(list).hasSize(100);
+    assertThat(list).isInOrder();
+  }
+
+  @Test
+  public void submitCallableThatReturns() throws Exception {
+    String ender = "Remember, the enemy's gate is down.";
+    Task<String> task = crashlyticsWorker.submit(() -> ender);
+
+    String result = Tasks.await(task);
+
+    assertThat(result).isEqualTo(ender);
+  }
+
+  @Test
+  public void submitCallableThatReturnsNull() throws Exception {
+    Task<String> task = crashlyticsWorker.submit(() -> null);
+
+    String result = Tasks.await(task);
+
+    assertThat(result).isNull();
+  }
+
+  @Test
+  public void submitCallableThatThrows() {
+    Task<Void> task =
+        crashlyticsWorker.submit(
+            () -> {
+              throw new Exception("I threw in the callable");
+            });
+
+    ExecutionException thrown = assertThrows(ExecutionException.class, () -> Tasks.await(task));
+
+    assertThat(thrown).hasCauseThat().hasMessageThat().isEqualTo("I threw in the callable");
+  }
+
+  @Test
+  public void submitCallableThatThrowsThenReturns() throws Exception {
+    Task<Void> throwingTask =
+        crashlyticsWorker.submit(
+            () -> {
+              throw new IOException();
+            });
+
+    assertThrows(ExecutionException.class, () -> Tasks.await(throwingTask));
+
+    String hiro =
+        "When you are wrestling for possession of a sword, the man with the handle always wins.";
+    Task<String> task = crashlyticsWorker.submitTask(() -> Tasks.forResult(hiro));
+
+    String result = Tasks.await(task);
+
+    assertThat(result).isEqualTo(hiro);
+  }
+
+  @Test
+  public void submitRunnable() throws Exception {
+    Task<Void> task = crashlyticsWorker.submit(() -> {});
+
+    Void result = Tasks.await(task);
+
+    // A Runnable does not return, so the task evaluates to null.
+    assertThat(result).isNull();
+  }
+
+  @Test
+  public void submitRunnableThatThrows() {
+    Task<Void> task =
+        crashlyticsWorker.submit(
+            (Runnable)
+                () -> {
+                  throw new RuntimeException("I threw in the runnable");
+                });
+
+    ExecutionException thrown = assertThrows(ExecutionException.class, () -> Tasks.await(task));
+
+    assertThat(thrown).hasCauseThat().hasMessageThat().isEqualTo("I threw in the runnable");
+  }
+
+  @Test
+  public void submitRunnableThatThrowsThenReturns() throws Exception {
+    Task<Void> thowingTask =
+        crashlyticsWorker.submit(
+            (Runnable)
+                () -> {
+                  throw new IllegalArgumentException();
+                });
+
+    assertThrows(ExecutionException.class, () -> Tasks.await(thowingTask));
+
+    Task<Void> task = crashlyticsWorker.submit(() -> {});
+
+    Void result = Tasks.await(task);
+
+    assertThat(result).isNull();
+  }
+
+  @Test
+  public void submitTaskThatReturns() throws Exception {
+    String skippy = "Think of the problem as an enemy, and defeat them in detail.";
+    Task<String> task = crashlyticsWorker.submitTask(() -> Tasks.forResult(skippy));
+
+    String result = Tasks.await(task);
+
+    assertThat(result).isEqualTo(skippy);
+  }
+
+  @Test
+  public void submitTaskThatReturnsNull() throws Exception {
+    Task<String> task = crashlyticsWorker.submitTask(() -> Tasks.forResult(null));
+
+    String result = Tasks.await(task);
+
+    assertThat(result).isNull();
+  }
+
+  @Test
+  public void submitTaskThatThrows() {
+    Task<String> task =
+        crashlyticsWorker.submitTask(
+            () -> Tasks.forException(new Exception("Thrown from a task.")));
+
+    ExecutionException thrown = assertThrows(ExecutionException.class, () -> Tasks.await(task));
+
+    assertThat(thrown).hasCauseThat().hasMessageThat().isEqualTo("Thrown from a task.");
+  }
+
+  @Test
+  public void submitTaskThatThrowsThenReturns() throws Exception {
+    crashlyticsWorker.submitTask(() -> Tasks.forException(new IllegalStateException()));
+    Task<String> task = crashlyticsWorker.submitTask(() -> Tasks.forResult("The Hail Mary"));
+
+    String result = Tasks.await(task);
+
+    assertThat(result).isEqualTo("The Hail Mary");
+  }
+
+  @Test
+  public void submitTaskThatCancels() {
+    Task<Void> task = crashlyticsWorker.submitTask(Tasks::forCanceled);
+
+    CancellationException thrown =
+        assertThrows(CancellationException.class, () -> Tasks.await(task));
+
+    assertThat(task.isCanceled()).isTrue();
+    assertThat(thrown).hasMessageThat().contains("Task is already canceled");
+  }
+
+  @Test
+  public void submitTaskThatCancelsThenReturns() throws Exception {
+    crashlyticsWorker.submitTask(Tasks::forCanceled);
+    Task<String> task = crashlyticsWorker.submitTask(() -> Tasks.forResult("Flying Dutchman"));
+
+    String result = Tasks.await(task);
+
+    assertThat(task.isCanceled()).isFalse();
+    assertThat(result).isEqualTo("Flying Dutchman");
+  }
+
+  @Test
+  public void submitTaskThatCancelsThenAwaitsThenReturns() throws Exception {
+    Task<?> cancelled = crashlyticsWorker.submitTask(Tasks::forCanceled);
+
+    // Await on the cancelled task to force the exception to propagate.
+    assertThrows(CancellationException.class, () -> Tasks.await(cancelled));
+
+    // Submit another task.
+    Task<String> task = crashlyticsWorker.submitTask(() -> Tasks.forResult("Valkyrie"));
+
+    String result = Tasks.await(task);
+
+    assertThat(cancelled.isCanceled()).isTrue();
+    assertThat(task.isCanceled()).isFalse();
+    assertThat(result).isEqualTo("Valkyrie");
+  }
+
+  @Test
+  public void submitTaskThatCancelsThenAwaitsThenCallable() throws Exception {
+    Task<?> cancelled = crashlyticsWorker.submitTask(Tasks::forCanceled);
+
+    // Await on the cancelled task to force the exception to propagate.
+    assertThrows(CancellationException.class, () -> Tasks.await(cancelled));
+
+    // Submit a simple callable.
+    Task<Boolean> task = crashlyticsWorker.submit(() -> true);
+
+    boolean result = Tasks.await(task);
+
+    assertThat(cancelled.isCanceled()).isTrue();
+    assertThat(task.isCanceled()).isFalse();
+    assertThat(result).isTrue();
+  }
+
+  @Test
+  public void submitTaskThatCancelsThenAwaitsThenRunnable() throws Exception {
+    Task<?> cancelled = crashlyticsWorker.submitTask(Tasks::forCanceled);
+
+    // Await on the cancelled task to force the exception to propagate.
+    assertThrows(CancellationException.class, () -> Tasks.await(cancelled));
+
+    // Submit an empty runnable.
+    Task<Void> task = crashlyticsWorker.submit(() -> {});
+
+    Void result = Tasks.await(task);
+
+    assertThat(cancelled.isCanceled()).isTrue();
+    assertThat(task.isCanceled()).isFalse();
+    assertThat(result).isNull();
+  }
+
+  @Test
+  public void submitTaskFromAnotherWorker() throws Exception {
+    Task<String> otherTask =
+        new CrashlyticsWorker(TestOnlyExecutors.blocking())
+            .submit(() -> "Dog's fine. Just sleeping.");
+
+    // This will not use a background thread while waiting for the task on blocking thread.
+    Task<String> task = crashlyticsWorker.submitTask(() -> otherTask);
+
+    String result = Tasks.await(task);
+    assertThat(result).isEqualTo("Dog's fine. Just sleeping.");
+  }
+
+  @Test
+  public void submitTaskFromAnotherWorkerThatThrows() throws Exception {
+    Task<?> otherTask =
+        new CrashlyticsWorker(TestOnlyExecutors.blocking())
+            .submitTask(() -> Tasks.forException(new IndexOutOfBoundsException()));
+
+    // Await on the throwing task to force the exception to propagate threw the local worker.
+    Task<?> task = crashlyticsWorker.submitTask(() -> otherTask);
+    assertThrows(ExecutionException.class, () -> Tasks.await(task));
+
+    // Submit another task to local worker to verify the chain did not break.
+    Task<Integer> localTask = crashlyticsWorker.submitTask(() -> Tasks.forResult(0x5f375a86));
+
+    int localResult = Tasks.await(localTask);
+
+    assertThat(otherTask.isSuccessful()).isFalse();
+    assertThat(localTask.isSuccessful()).isTrue();
+    assertThat(localResult).isEqualTo(0x5f375a86);
+  }
+
+  @Test
+  public void submitTaskFromAnotherWorkerThatCancels() throws Exception {
+    Task<?> otherCancelled =
+        new CrashlyticsWorker(TestOnlyExecutors.blocking()).submitTask(Tasks::forCanceled);
+
+    // Await on the cancelled task to force the exception to propagate threw the local worker.
+    Task<?> task = crashlyticsWorker.submitTask(() -> otherCancelled);
+    assertThrows(CancellationException.class, () -> Tasks.await(task));
+
+    // Submit another task to local worker to verify the chain did not break.
+    Task<Long> localTask = crashlyticsWorker.submitTask(() -> Tasks.forResult(0x5fe6eb50c7b537a9L));
+
+    long localResult = Tasks.await(localTask);
+
+    assertThat(otherCancelled.isCanceled()).isTrue();
+    assertThat(localTask.isCanceled()).isFalse();
+    assertThat(localResult).isEqualTo(0x5fe6eb50c7b537a9L);
+  }
+
+  @Test
+  public void submitTaskFromAnotherWorkerDoesNotUseLocalThreads() throws Exception {
+    // Setup a "local" worker.
+    ThreadPoolExecutor localExecutor = (ThreadPoolExecutor) Executors.newFixedThreadPool(4);
+    CrashlyticsWorker localWorker = new CrashlyticsWorker(localExecutor);
+
+    // Use a task off crashlyticsWorker to represent an other task.
+    Task<Integer> otherTask =
+        crashlyticsWorker.submit(
+            () -> {
+              sleep(30);
+              return localExecutor.getActiveCount();
+            });
+
+    // No active threads yet.
+    assertThat(localExecutor.getActiveCount()).isEqualTo(0);
+
+    // 1 active thread when doing a local task.
+    assertThat(Tasks.await(localWorker.submit(localExecutor::getActiveCount))).isEqualTo(1);
+
+    // 0 active local threads when waiting for other task.
+    // Waiting for a task from another worker does not block a local thread.
+    assertThat(Tasks.await(localWorker.submitTask(() -> otherTask))).isEqualTo(0);
+
+    // 1 active thread when doing a task.
+    assertThat(Tasks.await(localWorker.submit(localExecutor::getActiveCount))).isEqualTo(1);
+
+    // No active threads after.
+    assertThat(localExecutor.getActiveCount()).isEqualTo(0);
+  }
+
+  @Test
+  public void submitTaskWhenThreadPoolFull() {
+    // Fill the underlying executor thread pool.
+    for (int i = 0; i < 10; i++) {
+      crashlyticsWorker.getExecutor().execute(() -> sleep(40));
+    }
+
+    Task<Integer> task = crashlyticsWorker.submitTask(() -> Tasks.forResult(42));
+
+    // The underlying thread pool is full with tasks that will take longer than this timeout.
+    assertThrows(TimeoutException.class, () -> Tasks.await(task, 30, TimeUnit.MILLISECONDS));
+  }
+
+  private static void sleep(long millis) {
+    try {
+      Thread.sleep(millis);
+    } catch (InterruptedException ex) {
+      Thread.currentThread().interrupt();
+    }
+  }
+}
diff --git a/firebase-crashlytics/src/main/java/com/google/firebase/crashlytics/internal/CrashlyticsWorker.java b/firebase-crashlytics/src/main/java/com/google/firebase/crashlytics/internal/CrashlyticsWorker.java
new file mode 100644
index 00000000000..876f018fc95
--- /dev/null
+++ b/firebase-crashlytics/src/main/java/com/google/firebase/crashlytics/internal/CrashlyticsWorker.java
@@ -0,0 +1,128 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.firebase.crashlytics.internal;
+
+import androidx.annotation.VisibleForTesting;
+import com.google.android.gms.tasks.Task;
+import com.google.android.gms.tasks.Tasks;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+
+/**
+ * Helper for executing tasks sequentially on the given executor service.
+ *
+ * <p>Work on the queue may block, or it may return a Task, such that the underlying thread may be
+ * re-used while the worker queue is still blocked.
+ *
+ * <p>Work enqueued on this worker will be run serially, regardless of the underlying executor.
+ * Therefore, workers on the queue should not add new work to the queue and then block on it, as
+ * that would create a deadlock. In such a case, the worker can return a Task that depends on the
+ * future work, and run the future work on the executor's thread, but not put it in the queue as its
+ * own worker.
+ *
+ * @hide
+ */
+public class CrashlyticsWorker {
+  private final ExecutorService executor;
+
+  private final Object tailLock = new Object();
+  private Task<?> tail = Tasks.forResult(null);
+
+  public CrashlyticsWorker(ExecutorService executor) {
+    this.executor = executor;
+  }
+
+  /** Returns the executor used by this worker. */
+  public ExecutorService getExecutor() {
+    return executor;
+  }
+
+  /**
+   * Submits a <code>Callable</code> task for asynchronous execution on the executor.
+   *
+   * <p>Returns a <code>Task</code> which will be resolved upon successful completion of the
+   * callable, or throws an <code>ExecutionException</code> if the callable throws an exception.
+   */
+  public <T> Task<T> submit(Callable<T> callable) {
+    synchronized (tailLock) {
+      // Do not propagate a cancellation.
+      if (tail.isCanceled()) {
+        tail = tail.continueWithTask(executor, task -> Tasks.forResult(null));
+      }
+      // Chain the new callable onto the queue's tail.
+      Task<T> result = tail.continueWith(executor, task -> callable.call());
+      tail = result;
+      return result;
+    }
+  }
+
+  /**
+   * Submits a <code>Runnable</code> task for asynchronous execution on the executor.
+   *
+   * <p>Returns a <code>Task</code> which will be resolved with null upon successful completion of
+   * the runnable, or throws an <code>ExecutionException</code> if the runnable throws an exception.
+   */
+  public Task<Void> submit(Runnable runnable) {
+    synchronized (tailLock) {
+      // Do not propagate a cancellation.
+      if (tail.isCanceled()) {
+        tail = tail.continueWithTask(executor, task -> Tasks.forResult(null));
+      }
+      // Chain the new runnable onto the queue's tail.
+      Task<Void> result =
+          tail.continueWith(
+              executor,
+              task -> {
+                runnable.run();
+                return null;
+              });
+      tail = result;
+      return result;
+    }
+  }
+
+  /**
+   * Submits a <code>Callable</code> <code>Task</code> for asynchronous execution on the executor.
+   *
+   * <p>This is useful for making the worker block on an asynchronous operation, while letting the
+   * underlying threads be re-used.
+   *
+   * <p>Returns a <code>Task</code> which will be resolved upon successful completion of the Task
+   * returned by the callable, throws an <code>ExecutionException</code> if the callable throws an
+   * exception, or throws a <code>CancellationException</code> if the task is cancelled.
+   */
+  public <T> Task<T> submitTask(Callable<Task<T>> callable) {
+    synchronized (tailLock) {
+      // Chain the new callable task onto the queue's tail, regardless of cancellation.
+      Task<T> result = tail.continueWithTask(executor, task -> callable.call());
+      tail = result;
+      return result;
+    }
+  }
+
+  /**
+   * Blocks until all current pending tasks have completed.
+   *
+   * <p>This is not a shutdown, this does not stop new tasks from being submitted to the queue.
+   */
+  @VisibleForTesting
+  public void await() throws ExecutionException, InterruptedException {
+    // Submit an empty runnable, and await on it.
+    Tasks.await(submit(() -> {}));
+  }
+}
diff --git a/firebase-crashlytics/src/main/java/com/google/firebase/crashlytics/internal/common/CrashlyticsBackgroundWorker.java b/firebase-crashlytics/src/main/java/com/google/firebase/crashlytics/internal/common/CrashlyticsBackgroundWorker.java
index c5dcd04e49b..f5812bb6a4c 100644
--- a/firebase-crashlytics/src/main/java/com/google/firebase/crashlytics/internal/common/CrashlyticsBackgroundWorker.java
+++ b/firebase-crashlytics/src/main/java/com/google/firebase/crashlytics/internal/common/CrashlyticsBackgroundWorker.java
@@ -32,8 +32,12 @@
  * that would create a deadlock. In such a case, the worker can return a Task that depends on the
  * future work, and run the future work on the executor's thread, but not put it in the queue as its
  * own worker.
+ *
+ * @deprecated Use the generic CrashlyticsWorker instead.
  */
+@Deprecated
 public class CrashlyticsBackgroundWorker {
+  // TODO(mrober): Clean this up after moving everything to the generic CrashlyticsWorker.
   private final Executor executor;
 
   private Task<Void> tail = Tasks.forResult(null);