diff --git a/Package.swift b/Package.swift
index 42cadc4d1..c9da27dbc 100644
--- a/Package.swift
+++ b/Package.swift
@@ -31,6 +31,10 @@ let package = Package(
     .package(url: "https://github.com/pointfreeco/swift-custom-dump", from: "1.3.2"),
     .package(url: "https://github.com/pointfreeco/swift-snapshot-testing", from: "1.17.0"),
     .package(url: "https://github.com/pointfreeco/xctest-dynamic-overlay", from: "1.2.2"),
+    .package(url: "https://github.com/cwalo/TUSKit.git", branch: "cwalo/response-body"),
+    // TODO: point to TUSKit once branch is merged
+//    .package(url: "https://github.com/tus/TUSKit.git", from: "3.6.0"),
+//    .package(path: "../TUSKit"), /
     .package(url: "https://github.com/WeTransfer/Mocker", from: "3.0.0"),
   ],
   targets: [
@@ -147,7 +151,10 @@ let package = Package(
     .target(
       name: "Storage",
       dependencies: [
-        "Helpers"
+        "Helpers",
+        "TUSKit",
+        // Local TUSKit testing
+//        .product(name: "TUSKit", package: "TUSKit"),
       ]
     ),
     .testTarget(
diff --git a/Sources/Storage/Resumable/ResumableClientStore.swift b/Sources/Storage/Resumable/ResumableClientStore.swift
new file mode 100644
index 000000000..d6268c4dc
--- /dev/null
+++ b/Sources/Storage/Resumable/ResumableClientStore.swift
@@ -0,0 +1,31 @@
+import ConcurrencyExtras
+import Foundation
+
+/// Creates and stores ResumableUploadClient instances by bucketId
+actor ResumableClientStore {
+  private let configuration: StorageClientConfiguration
+
+  var clients = LockIsolated<[String: ResumableUploadClient]>([:])
+
+  init(configuration: StorageClientConfiguration) {
+    self.configuration = configuration
+  }
+
+  func getOrCreateClient(for bucketId: String) throws -> ResumableUploadClient {
+    if let client = clients.value[bucketId] {
+      return client
+    } else {
+      let client = try ResumableUploadClient(bucketId: bucketId, configuration: configuration)
+      clients.withValue { $0[bucketId] = client }
+      return client
+    }
+  }
+
+  func removeClient(for bucketId: String) {
+    clients.withValue { _ = $0.removeValue(forKey: bucketId) }
+  }
+
+  func removeAllClients() {
+    clients.setValue([:])
+  }
+}
diff --git a/Sources/Storage/Resumable/ResumableUpload+Status.swift b/Sources/Storage/Resumable/ResumableUpload+Status.swift
new file mode 100644
index 000000000..3186a497f
--- /dev/null
+++ b/Sources/Storage/Resumable/ResumableUpload+Status.swift
@@ -0,0 +1,36 @@
+import Foundation
+
+extension ResumableUpload {
+  public enum Status: Sendable, Equatable {
+    case queued(UUID)
+    case started(UUID)
+    case progress(UUID, uploaded: Int, total: Int)
+    case finished(UUID)
+    case cancelled(UUID)
+    case failed(UUID, any Error)
+    case fileError(any Error)
+    case clientError(any Error)
+
+    // TODO: more robust equatable implementation
+    public static func == (lhs: Status, rhs: Status) -> Bool {
+      switch (lhs, rhs) {
+      case (.queued(let lhsId), .queued(let rhsId)):
+        return lhsId == rhsId
+      case (.started(let lhsId), .started(let rhsId)):
+        return lhsId == rhsId
+      case (.progress(let lhsId, _, _), .progress(let rhsId, _, _)):
+        return lhsId == rhsId
+      case (.finished(let lhsId), .finished(let rhsId)):
+        return lhsId == rhsId
+      case (.cancelled(let lhsId), .cancelled(let rhsId)):
+        return lhsId == rhsId
+      case (.fileError(let lhsError), .fileError(let rhsError)):
+        return lhsError.localizedDescription == rhsError.localizedDescription
+      case (.clientError(let lhsError), .clientError(let rhsError)):
+        return lhsError.localizedDescription == rhsError.localizedDescription
+      default:
+        return false
+      }
+    }
+  }
+}
diff --git a/Sources/Storage/Resumable/ResumableUpload.swift b/Sources/Storage/Resumable/ResumableUpload.swift
new file mode 100644
index 000000000..6accaccc9
--- /dev/null
+++ b/Sources/Storage/Resumable/ResumableUpload.swift
@@ -0,0 +1,62 @@
+import ConcurrencyExtras
+import Foundation
+
+/// An instance of a ResumableUpload
+///
+/// Consumers can maintain a reference to the upload to observe its status, pause, and resume the upload
+public final class ResumableUpload: @unchecked Sendable {
+  public let id: UUID
+  public let context: [String: String]?
+
+  weak var client: ResumableUploadClient?
+  var statuses = LockIsolated<[Status]>([])
+  var continuations = LockIsolated<[UUID: AsyncStream<Status>.Continuation]>([:])
+
+  init(id: UUID, context: [String: String]?, client: ResumableUploadClient) {
+    self.id = id
+    self.context = context
+    self.client = client
+    self.statuses.setValue([.queued(id)])
+  }
+
+  func send(_ status: Status) {
+    statuses.withValue { $0.append(status) }
+    let currentContinuations = continuations.value
+    currentContinuations.values.forEach { $0.yield(status) }
+  }
+
+  func finish() {
+    let currentContinuations = continuations.value
+    continuations.setValue([:])
+    currentContinuations.values.forEach { $0.finish() }
+  }
+
+  public func currentStatus() -> Status {
+    statuses.value.last ?? .queued(id)
+  }
+
+  public func status() -> AsyncStream<Status> {
+    AsyncStream { continuation in
+      let streamID = UUID()
+
+      // Replay the last status
+      if let status = self.statuses.value.last {
+        continuation.yield(status)
+      }
+
+      continuations.withValue { $0[streamID] = continuation }
+      continuation.onTermination = { @Sendable _ in
+        self.continuations.withValue { _ = $0.removeValue(forKey: streamID) }
+      }
+    }
+  }
+
+  public func pause() throws {
+    try client?.pause(id: id)
+  }
+
+  public func resume() throws -> Bool {
+    guard let client else { return false }
+    return try client.resume(id: id)
+  }
+}
diff --git a/Sources/Storage/Resumable/ResumableUploadApi.swift b/Sources/Storage/Resumable/ResumableUploadApi.swift
new file mode 100644
index 000000000..7edd34bf3
--- /dev/null
+++ b/Sources/Storage/Resumable/ResumableUploadApi.swift
@@ -0,0 +1,73 @@
+import ConcurrencyExtras
+import Foundation
+import HTTPTypes
+import TUSKit
+
+/// Supabase Resumable Upload API
+public class ResumableUploadApi: StorageApi, @unchecked Sendable {
+  let bucketId: String
+  let clientStore: ResumableClientStore
+
+  init(bucketId: String, configuration: StorageClientConfiguration, clientStore: ResumableClientStore) {
+    self.bucketId = bucketId
+    self.clientStore = clientStore
+    super.init(configuration: configuration)
+  }
+
+  public func upload(file: URL, to path: String, options: FileOptions = .init()) async throws -> ResumableUpload {
+    let client = try await clientStore.getOrCreateClient(for: bucketId)
+    let upload = try client.uploadFile(filePath: file, path: path, options: options)
+    return upload
+  }
+
+  public func upload(data: Data, to path: String, pathExtension: String? = nil, options: FileOptions = .init()) async throws -> ResumableUpload {
+    let client = try await clientStore.getOrCreateClient(for: bucketId)
+    let upload = try client.upload(data: data, path: path, pathExtension: pathExtension, options: options)
+    return upload
+  }
+
+  public func pauseUpload(id: UUID) async throws {
+    let client = try await clientStore.getOrCreateClient(for: bucketId)
+    try client.pause(id: id)
+  }
+
+  public func pauseAllUploads() async throws {
+    let client = try await clientStore.getOrCreateClient(for: bucketId)
+    try client.pauseAllUploads()
+  }
+
+  public func resumeUpload(id: UUID) async throws -> Bool {
+    let client = try await clientStore.getOrCreateClient(for: bucketId)
+    return try client.resume(id: id)
+  }
+
+  public func retryUpload(id: UUID) async throws -> Bool {
+    let client = try await clientStore.getOrCreateClient(for: bucketId)
+    return try client.retry(id: id)
+  }
+
+  public func resumeAllUploads() async throws {
+    let client = try await clientStore.getOrCreateClient(for: bucketId)
+    try client.resumeAllUploads()
+  }
+
+  public func cancelUpload(id: UUID) async throws {
+    let client = try await clientStore.getOrCreateClient(for: bucketId)
+    try client.cancel(id: id)
+  }
+
+  public func cancelAllUploads() async throws {
+    let client = try await clientStore.getOrCreateClient(for: bucketId)
+    try client.cancelAllUploads()
+  }
+
+  public func getUploadStatus(id: UUID) async throws -> ResumableUpload.Status? {
+    let client = try await clientStore.getOrCreateClient(for: bucketId)
+    return client.status(id: id)
+  }
+
+  public func getUpload(id: UUID) async throws -> ResumableUpload? {
+    let client = try await clientStore.getOrCreateClient(for: bucketId)
+    return client.upload(for: id)
+  }
+}
diff --git a/Sources/Storage/Resumable/ResumableUploadClient.swift b/Sources/Storage/Resumable/ResumableUploadClient.swift
new file mode 100644
index 000000000..d1f4bbf37
--- /dev/null
+++ b/Sources/Storage/Resumable/ResumableUploadClient.swift
@@ -0,0 +1,218 @@
+import ConcurrencyExtras
+import Foundation
+import TUSKit
+
+/// A wrapper around TUSClient
+///
+/// One client per bucket
+final class ResumableUploadClient: @unchecked Sendable {
+  let client: TUSClient
+  let bucketId: String
+  let url: URL
+  let configuration: StorageClientConfiguration
+
+  var activeUploads = LockIsolated<[UUID: ResumableUpload]>([:])
+
+  // Track finished state if status is requested without a reference to a ResumableUpload
+  var finishedUploads = LockIsolated<Set<UUID>>([])
+
+  deinit {
+    print("ResumableUploadClient deinit")
+  }
+
+  init(
+    bucketId: String,
+    configuration: StorageClientConfiguration
+  ) throws {
+    self.bucketId = bucketId
+    self.configuration = configuration
+    self.url = configuration.url.appendingPathComponent("/upload/resumable")
+
+    let storageDirectory = Self.storageDirectory(for: bucketId)
+
+    let client = try TUSClient(
+      server: url,
+      sessionIdentifier: bucketId,
+      sessionConfiguration: configuration.resumableSessionConfiguration,
+      storageDirectory: storageDirectory
+    )
+
+    self.client = client
+    client.delegate = self
+  }
+
+  static func storageDirectory(for bucketId: String) -> URL {
+    FileManager.default
+      .urls(for: .documentDirectory, in: .userDomainMask)[0]
+      .appendingPathComponent("TUS/\(bucketId)")
+  }
+
+  func uploadFile(
+    filePath: URL,
+    path: String,
+    options: FileOptions = .init()
+  ) throws -> ResumableUpload {
+
+    let uploadURL = url
+    var headers = configuration.headers
+    headers["x-upsert"] = options.upsert ? "true" : "false"
+
+    let context: [String: String] = [
+      "bucketName": bucketId,
+      "objectName": path,
+      "contentType": options.contentType ?? mimeType(forPathExtension: filePath.pathExtension)
+    ]
+
+    // TODO: resume stored upload and check if there's already an active upload
+
+    let id = try client.uploadFileAt(
+      filePath: filePath,
+      uploadURL: uploadURL,
+      customHeaders: headers,
+      context: context
+    )
+
+    let upload = ResumableUpload(id: id, context: context, client: self)
+    activeUploads.withValue {
+        $0[id] = upload
+    }
+
+    return upload
+  }
+
+  func upload(
+    data: Data,
+    path: String,
+    pathExtension: String? = nil,
+    options: FileOptions = .init()
+  ) throws -> ResumableUpload {
+
+    let uploadURL = url
+    var headers = configuration.headers
+    headers["x-upsert"] = options.upsert ? "true" : "false"
+
+    var context: [String: String] = [
+      "bucketName": bucketId,
+      "objectName": path,
+    ]
+
+    if let contentType = pathExtension ?? options.contentType {
+      context["contentType"] = contentType
+    }
+
+    // TODO: check if there's already an active upload and resume a stored upload that has not been created
+    let id = try client.upload(
+      data: data,
+      uploadURL: uploadURL,
+      customHeaders: headers,
+      context: context
+    )
+
+    let upload = ResumableUpload(id: id, context: context, client: self)
+    activeUploads.withValue {
+        $0[id] = upload
+    }
+
+    return upload
+  }
+
+  func status(id: UUID) -> ResumableUpload.Status? {
+    if let activeUpload = activeUploads.value[id] {
+      return activeUpload.currentStatus()
+    } else if finishedUploads.value.contains(id) {
+      return .finished(id)
+    } else {
+      return nil
+    }
+
+    // TODO: check TUSClient if we don't have an active upload stored
+  }
+
+  func upload(for id: UUID) -> ResumableUpload? {
+    activeUploads.value[id]
+  }
+
+  func pause(id: UUID) throws {
+    try client.cancel(id: id)
+  }
+
+  func pauseAllUploads() throws {
+    client.stopAndCancelAll()
+  }
+
+  func resume(id: UUID) throws -> Bool {
+    return try client.resume(id: id)
+  }
+
+  func resumeAllUploads() throws {
+    let storedUploads = client.start()
+    activeUploads.withValue {
+      for (id, context) in storedUploads {
+        // Ensure we don't overwrite an upload that is created in `didStartUpload`
+        if $0.keys.contains(id) { continue }
+        $0[id] = ResumableUpload(id: id, context: context, client: self)
+      }
+    }
+  }
+
+  func retry(id: UUID) throws -> Bool {
+    return try client.retry(id: id)
+  }
+
+  func cancel(id: UUID) throws {
+    try client.cancel(id: id)
+    try client.removeCacheFor(id: id)
+  }
+
+  func cancelAllUploads() throws {
+    try client.reset()
+  }
+}
+
+extension ResumableUploadClient: TUSClientDelegate {
+  func didStartUpload(id: UUID, context: [String: String]?, client: TUSClient) {
+    if let upload = activeUploads.value[id] {
+      upload.send(.started(id))
+    } else {
+      // If an upload was resumed and it's not stored, create one
+      let upload = ResumableUpload(id: id, context: context, client: self)
+      activeUploads.withValue { $0[id] = upload }
+      upload.send(.started(id))
+    }
+  }
+
+  func progressFor(id: UUID, context: [String: String]?, bytesUploaded: Int, totalBytes: Int, client: TUSClient) {
+    if let upload = activeUploads.value[id] {
+      upload.send(.progress(id, uploaded: bytesUploaded, total: totalBytes))
+    }
+  }
+
+  func didFinishUpload(id: UUID, url: URL, context: [String: String]?, client: TUSClient) {
+    _ = finishedUploads.withValue { $0.insert(id) }
+
+    if let upload = activeUploads.value[id] {
+      upload.send(.finished(id))
+      upload.finish()
+      activeUploads.withValue { _ = $0.removeValue(forKey: id) }
+    }
+  }
+
+  func uploadFailed(id: UUID, error: any Error, context: [String: String]?, client: TUSClient) {
+    if let upload = activeUploads.value[id] {
+      upload.send(.failed(id, error))
+      upload.finish()
+      // TODO: not sure if the upload should be removed if it fails
+//      activeUploads.withValue { _ = $0.removeValue(forKey: id) }
+    }
+  }
+
+  func fileError(error: TUSClientError, client: TUSClient) {
+      // TODO: emit file error
+//    onFileError?(error)
+  }
+
+  func totalProgress(bytesUploaded: Int, totalBytes: Int, client: TUSClient) {
+      // TODO: emit total progress (for all upload)
+//    onTotalProgress?(bytesUploaded, totalBytes)
+  }
+}
diff --git a/Sources/Storage/StorageFileApi.swift b/Sources/Storage/StorageFileApi.swift
index 5ec49be97..df865fd68 100644
--- a/Sources/Storage/StorageFileApi.swift
+++ b/Sources/Storage/StorageFileApi.swift
@@ -59,8 +59,11 @@ public class StorageFileApi: StorageApi, @unchecked Sendable {
   /// The bucket id to operate on.
   let bucketId: String
 
-  init(bucketId: String, configuration: StorageClientConfiguration) {
+  public let resumable: ResumableUploadApi
+
+  init(bucketId: String, configuration: StorageClientConfiguration, clientStore: ResumableClientStore) {
     self.bucketId = bucketId
+    self.resumable = .init(bucketId: bucketId, configuration: configuration, clientStore: clientStore)
     super.init(configuration: configuration)
   }
 
diff --git a/Sources/Storage/SupabaseStorage.swift b/Sources/Storage/SupabaseStorage.swift
index ba043c8b8..31a2e6c1a 100644
--- a/Sources/Storage/SupabaseStorage.swift
+++ b/Sources/Storage/SupabaseStorage.swift
@@ -1,4 +1,5 @@
 import Foundation
+import ConcurrencyExtras
 
 public struct StorageClientConfiguration: Sendable {
   public var url: URL
@@ -6,6 +7,7 @@ public struct StorageClientConfiguration: Sendable {
   public let encoder: JSONEncoder
   public let decoder: JSONDecoder
   public let session: StorageHTTPSession
+  public let resumableSessionConfiguration: URLSessionConfiguration
   public let logger: (any SupabaseLogger)?
   public let useNewHostname: Bool
 
@@ -15,6 +17,7 @@ public struct StorageClientConfiguration: Sendable {
     encoder: JSONEncoder = .defaultStorageEncoder,
     decoder: JSONDecoder = .defaultStorageDecoder,
     session: StorageHTTPSession = .init(),
+    resumableSessionConfiguration: URLSessionConfiguration = .background(withIdentifier: "com.supabase.storage.resumable"),
     logger: (any SupabaseLogger)? = nil,
     useNewHostname: Bool = false
   ) {
@@ -23,16 +26,26 @@ public struct StorageClientConfiguration: Sendable {
     self.encoder = encoder
     self.decoder = decoder
     self.session = session
+    self.resumableSessionConfiguration = resumableSessionConfiguration
     self.logger = logger
     self.useNewHostname = useNewHostname
   }
 }
 
 public class SupabaseStorageClient: StorageBucketApi, @unchecked Sendable {
+  private let resumableStore = LockIsolated<ResumableClientStore?>(nil)
+
   /// Perform file operation in a bucket.
   /// - Parameter id: The bucket id to operate on.
   /// - Returns: StorageFileApi object
   public func from(_ id: String) -> StorageFileApi {
-    StorageFileApi(bucketId: id, configuration: configuration)
+    let clientStore = resumableStore.withValue {
+      if $0 == nil {
+        $0 = ResumableClientStore(configuration: configuration)
+      }
+      return $0!
+    }
+
+    return StorageFileApi(bucketId: id, configuration: configuration, clientStore: clientStore)
   }
 }
diff --git a/Tests/StorageTests/ResumableTests/ResumableClientStoreTests.swift b/Tests/StorageTests/ResumableTests/ResumableClientStoreTests.swift
new file mode 100644
index 000000000..0bc7819be
--- /dev/null
+++ b/Tests/StorageTests/ResumableTests/ResumableClientStoreTests.swift
@@ -0,0 +1,59 @@
+import ConcurrencyExtras
+import XCTest
+
+@testable import Storage
+
+final class ResumableClientStoreTests: XCTestCase {
+  var storage: SupabaseStorageClient!
+
+  override func setUp() {
+    super.setUp()
+
+    storage = SupabaseStorageClient.test(
+      supabaseURL: "http://localhost:54321/storage/v1",
+      apiKey: "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZS1kZW1vIiwicm9sZSI6ImFub24iLCJleHAiOjE5ODM4MTI5OTZ9.CRXP1A7WOeoJeXxjNni43kdQwgnWNReilDMblYTn_I0"
+    )
+  }
+
+  func testInitializeStore() async throws {
+    let api = storage.from("bucket").resumable
+    let store = api.clientStore
+    let clients = await store.clients.value
+    XCTAssertEqual(clients.values.count, 0)
+  }
+
+  func testCreateClient() async throws {
+    let api = storage.from("bucket").resumable
+    let store = api.clientStore
+    let client = try await store.getOrCreateClient(for: "bucket")
+    let clients = await store.clients.value
+    XCTAssertEqual(clients.values.count, 1)
+
+    let clientFromStore = await store.clients.value["bucket"]
+    XCTAssertNotNil(clientFromStore)
+    XCTAssertEqual(clientFromStore!.bucketId, client.bucketId)
+  }
+
+  func testRemoveClient() async throws {
+    let api = storage.from("bucket").resumable
+    let store = api.clientStore
+    _ = try await store.getOrCreateClient(for: "bucket")
+    var clients = await store.clients.value
+    XCTAssertEqual(clients.values.count, 1)
+    await store.removeClient(for: "bucket")
+    clients = await store.clients.value
+    XCTAssertEqual(clients.values.count, 0)
+  }
+
+  func testRemoveAllClients() async throws {
+    let api = storage.from("bucket").resumable
+    let store = api.clientStore
+    _ = try await store.getOrCreateClient(for: "bucket")
+    _ = try await store.getOrCreateClient(for: "bucket1")
+    var clients = await store.clients.value
+    XCTAssertEqual(clients.values.count, 2)
+    await store.removeAllClients()
+    clients = await store.clients.value
+    XCTAssertEqual(clients.values.count, 0)
+  }
+}
diff --git a/Tests/StorageTests/ResumableTests/ResumableUploadAPITests.swift b/Tests/StorageTests/ResumableTests/ResumableUploadAPITests.swift
new file mode 100644
index 000000000..df97cc137
--- /dev/null
+++ b/Tests/StorageTests/ResumableTests/ResumableUploadAPITests.swift
@@ -0,0 +1,302 @@
+import InlineSnapshotTesting
+import Mocker
+import TestHelpers
+import TUSKit
+import XCTest
+import ConcurrencyExtras
+
+#if canImport(FoundationNetworking)
+  import FoundationNetworking
+#endif
+
+@testable import Storage
+
+final class ResumableUploadAPITests: XCTestCase {
+  var storage: SupabaseStorageClient!
+
+  override func setUp() {
+    super.setUp()
+
+    storage = SupabaseStorageClient.test(
+      supabaseURL: "http://localhost:54321/storage/v1",
+      apiKey: "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZS1kZW1vIiwicm9sZSI6ImFub24iLCJleHAiOjE5ODM4MTI5OTZ9.CRXP1A7WOeoJeXxjNni43kdQwgnWNReilDMblYTn_I0"
+    )
+  }
+
+  override func tearDown() {
+    super.tearDown()
+    removeStoredUploads()
+  }
+
+  func removeStoredUploads() {
+    let storageDirectory = ResumableUploadClient.storageDirectory(for: "bucket")
+    try? FileManager.default.removeItem(at: storageDirectory)
+  }
+
+  func createUpload(removeExistingFile: Bool = true) async throws -> ResumableUpload {
+    let testFileURL = Bundle.module.url(forResource: "file", withExtension: "txt")!
+
+    let api = storage.from("bucket")
+
+    // Remove existing file
+    if removeExistingFile {
+      try await api.remove(paths: ["file.txt"])
+    }
+
+    let resumable = api.resumable
+    let upload = try await resumable.upload(file: testFileURL, to: "file.txt")
+    return upload
+  }
+
+  func testCreateApi() throws {
+    let api = storage.from("bucket").resumable
+    XCTAssertEqual(api.bucketId, "bucket")
+    XCTAssertEqual(api.configuration.url, storage.configuration.url)
+  }
+
+  func testUploadFileContext() async throws {
+    let testFileURL = Bundle.module.url(forResource: "file", withExtension: "txt")!
+
+    let api = storage.from("bucket").resumable
+    let upload = try await api.upload(file: testFileURL, to: "file.txt")
+    XCTAssertEqual(upload.context, [
+      "objectName": "file.txt",
+      "contentType": "text/plain",
+      "bucketName": "bucket",
+    ])
+  }
+
+  func testUploadFileStatus() async throws {
+    let testFileURL = Bundle.module.url(forResource: "file", withExtension: "txt")!
+    
+    let api = storage.from("bucket")
+
+    // Remove existing file
+    try await api.remove(paths: ["file.txt"])
+
+    let resumable = api.resumable
+    let upload = try await resumable.upload(file: testFileURL, to: "file.txt")
+
+    var statuses = [ResumableUpload.Status]()
+    for await status in upload.status() {
+      statuses.append(status)
+    }
+
+    XCTAssertTrue(statuses.contains(where: { $0 == .started(upload.id) }))
+    XCTAssertTrue(statuses.contains(where: { $0 == .finished(upload.id) }))
+  }
+
+  func testUploadDuplicateFileSucceedsWithUpsert() async throws {
+    let testFileURL = Bundle.module.url(forResource: "file", withExtension: "txt")!
+
+    let api = storage.from("bucket")
+
+    // Remove existing file
+    try await api.remove(paths: ["file.txt"])
+
+    let resumable = api.resumable
+    let upload = try await resumable.upload(file: testFileURL, to: "file.txt")
+
+    var statuses = [ResumableUpload.Status]()
+    for await status in upload.status() {
+      statuses.append(status)
+    }
+
+    XCTAssertTrue(statuses.contains(where: { $0 == .finished(upload.id) }))
+
+    let upload2 = try await resumable.upload(file: testFileURL, to: "file.txt", options: .init(upsert: true))
+    var statuses2 = [ResumableUpload.Status]()
+    for await status in upload2.status() {
+      statuses2.append(status)
+    }
+
+    XCTAssertTrue(statuses2.contains(where: { $0 == .finished(upload2.id) }))
+  }
+
+  func testUploadDuplicateFileFailsWithoutUpsert() async throws {
+    let testFileURL = Bundle.module.url(forResource: "file", withExtension: "txt")!
+
+    let api = storage.from("bucket")
+
+    // Remove existing file
+    try await api.remove(paths: ["file.txt"])
+
+    let resumable = api.resumable
+    let upload = try await resumable.upload(file: testFileURL, to: "file.txt")
+
+    var statuses = [ResumableUpload.Status]()
+    for await status in upload.status() {
+      statuses.append(status)
+    }
+
+    XCTAssertTrue(statuses.contains(where: { $0 == .finished(upload.id) }))
+
+    let upload2 = try await resumable.upload(file: testFileURL, to: "file.txt")
+    var statuses2 = [ResumableUpload.Status]()
+    for await status in upload2.status() {
+      statuses2.append(status)
+    }
+
+    // TODO: check that error == couldNotCreateFileOnServer
+    XCTAssertTrue(statuses2.contains(where: {
+      if case let .failed(id, _) = $0, id == upload2.id {
+        true
+      } else {
+        false
+      }
+    }))
+  }
+
+  func testPauseAndResumeUpload() async throws {
+    let testFileURL = Bundle.module.url(forResource: "file", withExtension: "txt")!
+
+    let api = storage.from("bucket")
+
+    // Remove existing file
+    try await api.remove(paths: ["file.txt"])
+
+    let resumable = api.resumable
+    let upload = try await resumable.upload(file: testFileURL, to: "file.txt")
+    for await status in upload.status() {
+      if status == .started(upload.id) {
+        try upload.pause()
+      }
+    }
+
+    let currentStatus = upload.currentStatus()
+    if case let .failed(id, error) = currentStatus {
+      XCTAssertEqual(id, upload.id)
+      XCTAssertEqual(error.localizedDescription, TUSClientError.taskCancelled.localizedDescription)
+    } else {
+      XCTFail()
+    }
+
+    let didResume = try upload.resume()
+    XCTAssertTrue(didResume)
+
+    var statuses = [ResumableUpload.Status]()
+    for await status in upload.status() {
+      statuses.append(status)
+    }
+
+    XCTAssertTrue(statuses.contains(where: { $0 == .finished(upload.id) }))
+  }
+
+  func testCanceledUploadCannotBeResumed() async throws {
+    let testFileURL = Bundle.module.url(forResource: "file", withExtension: "txt")!
+
+    let api = storage.from("bucket")
+
+    // Remove existing file
+    try await api.remove(paths: ["file.txt"])
+
+    let resumable = api.resumable
+    let upload = try await resumable.upload(file: testFileURL, to: "file.txt")
+    for await status in upload.status() {
+      if status == .started(upload.id) {
+        // Pause the upload to simulate a transient failure
+        try upload.pause()
+      }
+    }
+
+    let currentStatus = upload.currentStatus()
+    if case let .failed(id, error) = currentStatus {
+      XCTAssertEqual(id, upload.id)
+      XCTAssertEqual(error.localizedDescription, TUSClientError.taskCancelled.localizedDescription)
+    } else {
+      XCTFail()
+    }
+
+    // While the upload is technically in the failed state when paused,
+    // it can be resumed unless the cache is cleared, which is what cancel does
+    try await resumable.cancelUpload(id: upload.id)
+
+    let didResume = try await resumable.resumeUpload(id: upload.id)
+    XCTAssertFalse(didResume)
+  }
+
+  func testGetCurrentUploadStatus() async throws {
+    let testFileURL = Bundle.module.url(forResource: "file", withExtension: "txt")!
+
+    let api = storage.from("bucket")
+
+    // Remove existing file
+    try await api.remove(paths: ["file.txt"])
+
+    let resumable = api.resumable
+    let upload = try await resumable.upload(file: testFileURL, to: "file.txt")
+    for await status in upload.status() {
+      let currentStatus = upload.currentStatus()
+      XCTAssertEqual(status, currentStatus)
+      let apiStatus = try await resumable.getUploadStatus(id: upload.id)
+      XCTAssertEqual(status, apiStatus)
+    }
+  }
+
+  func testResumeAllUploadsOnReinit() async throws {
+    let testFileURL = Bundle.module.self.url(forResource: "file", withExtension: "txt")!
+    let api = storage.from("bucket")
+    let resumable = api.resumable
+
+    try await api.remove(paths: ["file.txt"])
+
+    var upload: ResumableUpload! = try await resumable.upload(file: testFileURL, to: "file.txt")
+    let id = upload.id
+
+    // make sure the weak client is deinit'd
+    upload = nil
+
+    // Simulate a
+    _ = try await resumable.pauseAllUploads()
+
+    // Remove existing client
+    await resumable.clientStore.removeClient(for: "bucket")
+
+    // Creates a new client on first method call
+    _ = try await api.resumable.resumeAllUploads()
+    upload = try await api.resumable.getUpload(id: id)
+    XCTAssertNotNil(upload)
+
+    var statuses: [ResumableUpload.Status] = []
+    for await status in upload.status() {
+      statuses.append(status)
+    }
+
+    XCTAssertTrue(statuses.contains(where: { $0 == .finished(upload.id) }))
+  }
+
+  func testRetryFailedUpload() async throws {
+    let testFileURL = Bundle.module.url(forResource: "file", withExtension: "txt")!
+
+    let api = storage.from("bucket")
+
+    // Remove existing file
+    try await api.remove(paths: ["file.txt"])
+
+    let resumable = api.resumable
+    let upload = try await resumable.upload(file: testFileURL, to: "file.txt")
+    for await status in upload.status() {
+      if status == .started(upload.id) {
+        try await api.resumable.cancelUpload(id: upload.id)
+      }
+    }
+
+    let currentStatus = upload.currentStatus()
+    if case let .failed(id, error) = currentStatus {
+      XCTAssertEqual(id, upload.id)
+      XCTAssertEqual(error.localizedDescription, TUSClientError.taskCancelled.localizedDescription)
+    } else {
+      XCTFail()
+    }
+
+    let didRetry = try await resumable.retryUpload(id: upload.id)
+    XCTAssertTrue(didRetry)
+
+    var statuses = [ResumableUpload.Status]()
+    for await status in upload.status() {
+      statuses.append(status)
+    }
+
+    XCTAssertTrue(statuses.contains(where: { $0 == .finished(upload.id) }))
+  }
+}
diff --git a/Tests/StorageTests/SupabaseStorageClient+Test.swift b/Tests/StorageTests/SupabaseStorageClient+Test.swift
index ac10137f8..d8c68ff87 100644
--- a/Tests/StorageTests/SupabaseStorageClient+Test.swift
+++ b/Tests/StorageTests/SupabaseStorageClient+Test.swift
@@ -23,6 +23,7 @@ extension SupabaseStorageClient {
           "X-Client-Info": "storage-swift/x.y.z",
         ],
         session: session,
+        resumableSessionConfiguration: .default,
         logger: nil
       )
     )