Skip to content

Writes and the offline queue

By default every insert, upsert, update and delete is saved to the mutation queue first, and sent at once when nothing of the same user waits ahead of it; otherwise it waits and is sent by a later pass, in the order it was made. A write returns a MutationOutcome, Applied with the server's answer or Queued, and every cached read shows it at once either way.

From the samples: OfflineTaskRepository.kt · TaskRepository.swift

/** Adds a task; [needsConnection] picks the write's strategy. */
override suspend fun addTask(
    title: String,
    notes: String?,
    needsConnection: Boolean,
): NewTaskOutcome {
    val row = newTaskRow(title, notes)
    val outcome =
        write(Tables.TASKS, TaskMutationOperation.Insert, title) {
            offline.from(Tables.TASKS).insert(row, mutationStrategyFor(needsConnection)) {
                select() // the stored row back, with the times the server set
            }
        }
    return when (outcome) {
        is MutationOutcome.Applied -> {
            val stored = outcome.result.decodeSingle<TaskRecord>()
            // An online-only write is never patched into the cache: the stored row goes there.
            if (needsConnection) offline.applyServerRecord(Tables.TASKS, stored)
            NewTaskOutcome.Saved(outcome.result.decodeSingle<TaskRow>().toTask(emptySet()))
        }
        is MutationOutcome.Queued -> NewTaskOutcome.SavedOnDevice(row.id) // sent when it can be
    }
}

/// Adds task `id` (a new one unless given) for the signed-in user, with the device's clock as
/// its creation time, and says what happened: the server stored it, at the `updated_at` it set
/// and gives back, or it waits in the queue on this device.
func addTask(id: String = DeviceID.make(), title: String, notes: String?, needsConnection: Bool = false) async throws -> NewTaskResult {
    let row = NewTaskRow(
        id: id,
        userID: try signedInUserID(),
        title: title,
        notes: notes,
        done: false,
        createdAt: .now
    )
    let outcome = try await write(taskTitle: title) {
        try await client.from(Tables.tasks)
            .insert(row, mutationStrategy: Self.mutationStrategy(needsConnection: needsConnection))
            .select()
            .single()
            .execute(StoredTaskRow.self)
    }
    switch outcome {
    case .applied(let result):
        // A write sent only straight to the server is never put into the device's lists:
        // the app puts in the row the server stored, as it would a row Realtime reports.
        if needsConnection { try await client.applyServerRecord(table: Tables.tasks, record: result.value) }
        return .saved(storedAt: result.value.updatedAt)
    case .queued(let pendingMutation):
        return .savedOnDevice(pendingMutationID: pendingMutation.id)
    }
}

The verbs are supabase-kt's, with its overloads and request DSL; Move to another task, on Storage caching, uses the update DSL. An update with a typed value, and one with a JSON body:

From the samples: OfflineTaskRepository.kt · TaskRepository.swift

/** Autosave: the title and notes as one typed value; null notes clear them on the server. */
override suspend fun updateTask(
    id: String,
    title: String,
    notes: String?,
): TaskMutationOutcome =
    write(Tables.TASKS, TaskMutationOperation.Update, title) {
            offline.from(Tables.TASKS).update(TaskTextPatch(title, notes)) {
                filter { eq(ID, id) }
            }
        }
        .toTaskMutationOutcome()


/** The done switch: the change as a JSON body, one column. */
override suspend fun setTaskDone(id: String, isDone: Boolean): TaskMutationOutcome =
    write(Tables.TASKS, TaskMutationOperation.Update, taskTitle = null) {
            offline.from(Tables.TASKS).update(buildJsonObject { put(DONE, isDone) }) {
                filter { eq(ID, id) }
            }
        }
        .toTaskMutationOutcome()

/// Sets a task's title and notes, as one typed patch; `nil` notes clears them on the server
/// too.
@discardableResult
func updateTaskText(id: String, title: String, notes: String?) async throws -> MutationOutcome<Void> {
    try await write(taskTitle: title) {
        try await client.from(Tables.tasks)
            .update(TaskTextPatch(title: title, notes: notes))
            .eq("id", value: id)
            .execute()
    }
}

/// Marks a task done, or open again, with the update's body written as JSON values.
@discardableResult
func setTaskDone(id: String, isDone: Bool) async throws -> MutationOutcome<Void> {
    try await write(taskTitle: nil) {
        try await client.from(Tables.tasks).update(["done": .bool(isDone)]).eq("id", value: id).execute()
    }
}

An upsert that asks for its row back, and a delete by filter:

From the samples: OfflineTaskRepository.kt · TaskRepository.swift

/** A copy of the task under a new id, written with `upsert`: a new key makes a new row. */
override suspend fun duplicateTask(id: String): NewTaskOutcome {
    val original =
        task(id).first { it.value != null }.value?.firstOrNull()
            ?: throw NotSavedOnDeviceException()
    val title = takeCharacters("${original.title} (copy)", TASK_TITLE_MAX_LENGTH)
    val copy = newTaskRow(title, original.notes, done = original.isDone)
    val outcome =
        write(Tables.TASKS, TaskMutationOperation.Upsert, title) {
            offline.from(Tables.TASKS).upsert(copy) { select() }
        }
    if (outcome.isQueued) return NewTaskOutcome.SavedOnDevice(copy.id) // no answer yet
    val stored = checkNotNull(outcome.resultOrNull()).decodeSingle<TaskRow>()
    return NewTaskOutcome.Saved(stored.toTask(pendingIds = emptySet()))
}


/** A task's photo files (online only), then its photo rows, then the task itself. */
override suspend fun deleteTask(id: String): TaskMutationOutcome = detached {
    val photos = cachedPhotosOf(id)
    // Files first, and only online: Storage writes are never queued, so offline they stay.
    if (canReachNetwork())
        deleteFiles(photos?.map { it.objectPath } ?: photoPathsOnServer(TASK_ID, id))
    // Then its photos still on this device: one whose upload has not returned yet gets no row.
    pendingPhotos.removeTask(id)
    // The server's cascade would delete the photo rows, but the cache never sees a cascade.
    if (photos == null || photos.isNotEmpty()) {
        write(Tables.TASK_PHOTOS, TaskMutationOperation.Delete, taskTitle = null) {
            offline.from(Tables.TASK_PHOTOS).delete { filter { eq(TASK_ID, id) } }
        }
    }
    write(Tables.TASKS, TaskMutationOperation.Delete, taskTitle = null) {
            offline.from(Tables.TASKS).delete { filter { eq(ID, id) } }
        }
        .toTaskMutationOutcome()
}

/// Makes the task's copy. The copy's id is made from the task's, so doing it again writes over
/// that copy rather than adding another: an upsert, merged by the table's key.
func duplicateTask(_ task: TaskItem) async throws -> MutationOutcome<Void> {
    let copy = NewTaskRow(
        id: DeviceID.derived(from: "copy of \(task.id)"),
        userID: task.userID,
        title: Wording.copyTitle(of: task.title),
        notes: task.notes,
        done: false,
        createdAt: .now
    )
    return try await write(taskTitle: copy.title) {
        try await client.from(Tables.tasks).upsert(copy).execute()
    }
}

try await write(taskTitle: nil) {
    try await client.from(Tables.tasks).delete().eq("id", value: id).execute()
}

Several rows in one write:

From the samples: OfflineTaskRepository.kt · TaskRepository.swift

/** Five tasks in one write: one request online, one queued write offline. */
override suspend fun addSampleTasks(): TaskMutationOutcome {
    val rows = SAMPLE_TITLES.map { title -> newTaskRow(title, notes = null) }
    return write(Tables.TASKS, TaskMutationOperation.Insert, rows.first().title) {
            offline.from(Tables.TASKS).insert(rows)
        }
        .toTaskMutationOutcome()
}


/**
 * The starter list, each task under an id made from the user's: imported twice, the second
 * changes nothing, since the server keeps the row it has for each id.
 */
override suspend fun importStarterList(): TaskMutationOutcome {
    val userId = signedInUserId()
    val rows = STARTER_TITLES.mapIndexed { index, title ->
        newTaskRow(title, null, id = starterId(userId, index))
    }
    return write(Tables.TASKS, TaskMutationOperation.Upsert, rows.first().title) {
            offline.from(Tables.TASKS).upsert(rows) {
                onConflict = ID // the column that decides a duplicate
                ignoreDuplicates = true // keep the stored row, edited or not
            }
        }
        .toTaskMutationOutcome()
}


/** A task with a blank title, as a JSON body the server's title check refuses. */
override suspend fun addTaskWithBlankTitle(): TaskMutationOutcome {
    val body = buildJsonArray {
        addJsonObject {
            put(ID, newId())
            put(USER_ID, signedInUserId())
            put(TITLE, "   ") // the server refuses a title with no text
            put(CREATED_AT, clock.now().toString())
        }
    }
    return write(Tables.TASKS, TaskMutationOperation.Insert, taskTitle = null) {
            offline.from(Tables.TASKS).insert(body)
        }
        .toTaskMutationOutcome()
}

/// Adds five sample tasks in one insert of a list of rows: one write in the queue, sent as one
/// request.
func addSampleTasks() async throws -> MutationOutcome<Void> {
    let userID = try signedInUserID()
    let now = Date.now
    let rows = SampleTasks.titles.enumerated().map { index, title in
        NewTaskRow(
            id: DeviceID.make(),
            userID: userID,
            title: title,
            notes: nil,
            done: false,
            // A second apart, so they list in the order written.
            createdAt: now - Double(SampleTasks.titles.count - index)
        )
    }
    return try await write(taskTitle: nil) {
        try await client.from(Tables.tasks).insert(rows).execute()
    }
}

/// Adds the starter tasks, once: their ids are made from the user's, so importing again finds
/// them on the server and changes nothing, however they were edited since. The rows are JSON
/// values; `onConflict` names the key and `ignoreDuplicates` keeps the server's rows.
func importStarterList() async throws -> MutationOutcome<Void> {
    let userID = try signedInUserID()
    let createdAt = Date.now.formatted(.iso8601)
    let rows: [OfflineJSON] = StarterList.titles.enumerated().map { index, title in
        // Every column the list reads: a queued row shows with only what it wrote.
        [
            "id": .string(DeviceID.derived(from: "starter \(index) of \(userID)")),
            "user_id": .string(userID),
            "title": .string(title),
            "notes": .null,
            "done": false,
            "created_at": .string(createdAt),
        ]
    }
    return try await write(taskTitle: nil) {
        try await client.from(Tables.tasks)
            .upsert(.array(rows), onConflict: "id", ignoreDuplicates: true)
            .execute()
    }
}

/// Adds a task whose title the server refuses, written as raw JSON, to show what a refused
/// write looks like: online it is refused at once; offline when it syncs.
@discardableResult
func queueInvalidTitle() async throws -> MutationOutcome<Void> {
    let row: OfflineJSON = [
        "id": .string(DeviceID.make()),
        "user_id": .string(try signedInUserID()),
        "title": .string(TaskRules.invalidTitle),
        "notes": .null,
        "done": false,
        "created_at": .string(Date.now.formatted(.iso8601)),
    ]
    return try await write(taskTitle: TaskRules.invalidTitle) {
        try await client.from(Tables.tasks).insert(row).execute()
    }
}

The two strategies

From the samples: OfflineTaskRepository.kt · TaskRepository.swift

/** The "Needs a connection" switch: queued when offline (the default), or sent now or never. */
private fun mutationStrategyFor(needsConnection: Boolean): MutationStrategy =
    if (needsConnection) MutationStrategy.OnlineOnly else MutationStrategy.QueueIfOffline


/** Deletes [userId]'s files, photo rows and tasks on the server, before that user is left. */
private suspend fun deleteEverythingOf(userId: String) {
    deleteFiles(photoPathsOnServer(USER_ID, userId))
    // Sent now or not at all: queued, they would outlive the user they are for.
    write(Tables.TASK_PHOTOS, TaskMutationOperation.Delete, taskTitle = null) {
        offline.from(Tables.TASK_PHOTOS).delete(
            mutationStrategy = MutationStrategy.OnlineOnly
        ) {
            filter { eq(USER_ID, userId) }
        }
    }
    write(Tables.TASKS, TaskMutationOperation.Delete, taskTitle = null) {
        offline.from(Tables.TASKS).delete(mutationStrategy = MutationStrategy.OnlineOnly) {
            filter { eq(USER_ID, userId) }
        }
    }
}

/// How New task sends its row: through the queue, so it is kept while offline (the default),
/// or straight to the server only, failing offline, when Needs a connection is on.
static func mutationStrategy(needsConnection: Bool) -> MutationStrategy {
    needsConnection ? .onlineOnly : .queueIfOffline
}

/// Deletes `userID`'s files, photo rows and tasks on the server, before that user is left.
/// The deletes are sent now or not at all: queued, they would outlive the user they are for.
private func deleteEverything(of userID: String) async throws {
    try await deleteFiles(photoPathsOnServer { $0.eq("user_id", value: userID) })
    // The files of its photos waiting to upload too: an upload whose answer was lost may have stored one.
    try await deleteFiles(pendingPhotos.photos.filter { $0.userID == userID }.map(\.objectPath))
    try await write(taskTitle: nil) {
        try await client.from(Tables.taskPhotos).delete(mutationStrategy: .onlineOnly).eq("user_id", value: userID).execute()
    }
    try await write(taskTitle: nil) {
        try await client.from(Tables.tasks).delete(mutationStrategy: .onlineOnly).eq("user_id", value: userID).execute()
    }
}
  • QueueIfOffline, the default, saves the write to the queue, so it keeps its place among the user's writes in every table, and keeps it there until the server has accepted or refused it.
  • OnlineOnly sends the write now and throws when it cannot. It is never queued and never patched into the cache, so it can reach the server before writes still queued, to the same rows or to a parent row it depends on. Use it for a write that makes no sense later, such as deleting an account's data.

What a write returns

MutationOutcome is a closed set, so a when over it stays exhaustive: Applied(result) holds the server's answer as supabase-kt returns it, and Queued(pendingMutation) the queued write, whose id identifies it from then on, on the rejected list too. isQueued says which, and resultOrNull() is the Applied result or null. A QueueIfOffline write returns:

When it is made Outcome
Nothing of the user waits ahead of it, the session is settled, and the server accepts it Applied, and the cached lists are patched with the rows the server returned, or with the rows sent when it returned none
Something of the user waits ahead of it, in any table, or the session is still restoring Queued, and a pass is started to send it
Network access is off (disableNetwork()) Queued; nothing is sent until access is back on
Sent at once, and it may not have run: no connection or no answer, a 503 or 520, an expired token that could not be refreshed, or PostgREST's schema cache not knowing the table or a column yet Queued. A later pass sends it: when the network comes back, the session settles, another write is made or the app syncs
Sent at once, and the server refused it with any other answer The server's exception is thrown, as supabase.from(…) would, and the write leaves the queue; it never reaches the rejected list, which holds only refusals found on replay (Conflicts and failures)
The caller is canceled while the write is on the wire It stays queued, since the server may have it, and a pass sends it

Writes the queue cannot hold

These are sent at once instead, as OnlineOnly writes are, except that a success still updates the cache. One that cannot be sent for want of a connection throws MutationNotQueueableException, whose cause is the connectivity failure and whose message names the kind:

  • an UPDATE or DELETE with no filter, or one the queue cannot record exactly: an or { } or and { } group, a negated filter, a filter on an embedded table's column, or a limit, order or offset (a queued UPDATE or DELETE carries filters on the table's own columns, joined as AND, with any of supabase-kt's operators);
  • a dry run (below);
  • a write to a schema other than the client's default (below; OnlineOnly sends it as any write);
  • a write whose user cannot be read from the stored session;
  • a write that was on its way into the queue while local data was cleared.

Two filters on one column

An update or a delete with two top-level filters on one column throws IllegalArgumentException under either strategy, online or offline, before anything is sent or queued (Coming from supabase-kt). One and { } group is the fix, and a group is a filter the queue cannot record, so such a write is sent at once: Try it makes one offline to show its MutationNotQueueableException.

From the samples: OfflineTaskRepository.kt · TaskRepository.swift

/**
 * This week's done tasks, deleted with one grouped filter. The queue cannot record a group, so
 * the write is sent at once, and refused with no connection rather than queued.
 */
override suspend fun deleteThisWeeksDoneTasks(): TaskMutationOutcome {
    val week = recentWindow(clock.now())
    return write(Tables.TASKS, TaskMutationOperation.Delete, taskTitle = null) {
            offline.from(Tables.TASKS).delete {
                filter {
                    eq(DONE, true)
                    and {
                        gte(CREATED_AT, week.from.toString())
                        lt(CREATED_AT, week.until.toString())
                    }
                }
            }
        }
        .toTaskMutationOutcome()
}

/// Deletes the done tasks created within `window`: two conditions on `created_at`, which the
/// client sends together in one group. The queue cannot record a group, so the delete is sent
/// at once rather than queued, and offline it fails with `mutationNotQueueable`.
func deleteDoneTasks(createdWithin window: DateInterval) async throws {
    try await client.from(Tables.tasks)
        .delete()
        .is("done", value: true)
        .gte("created_at", value: window.start)
        .lt("created_at", value: window.end)
        .execute()
}

Swift sends two conditions on one column together, in one and group, so nothing is refused.

Dry runs

A dry run (Prefer: tx=rollback: supabase-kt 3.8's dryRun(), or the header added by hand on 3.6) is never queued, since a replay would run it for real: it is sent at once, and offline throws MutationNotQueueableException. PostgREST rolls it back only when the project lets a request choose (db-tx-end set to commit-allow-override or rollback-allow-override), which a hosted Supabase project does not by default, so there a dry run commits: the library reads Preference-Applied, applies such a write like any other, patches its rows into the cache and logs a warning once per client. Check a dry run against a project that allows the override, or do not rely on it.

val outcome =
    supabase.offline.from("todos").insert(todo) {
        headers.append("Prefer", "tx=rollback") // sent at once, never queued
    }

Upserts and ids

A queued write keeps onConflict, ignoreDuplicates and defaultToNull for its replay. The cache merges an upsert by its onConflict columns or the table's key (id unless configured) (Optimistic overlay).

Make ids on the device: such a row can be shown, opened and edited before it reaches the server, and is never inserted twice. A keyed INSERT (its rows carry the table's primary key) whose earlier send may have reached the server is replayed as an upsert that ignores duplicates, and SyncReport.alreadyOnServerCount counts such rows. An INSERT whose key the server makes is inserted twice when the answer to its first send was lost; it can show twice in a read of its table answered while it is on the wire and, unless it asks for its rows back (select()), in that query's cached rows, across restarts, until they are fetched again; what a replay overwrites is on Conflicts and failures. A queued row carries only the columns it wrote until the list is fetched again (Optimistic overlay).

Another schema and the request lambda

The queue replays to the client's default schema only, so a QueueIfOffline write to another schema is sent at once and does not wait for writes queued to the default one; offline it throws a MutationNotQueueableException naming the schema, and online it marks the cached queries of that schema's table stale. Name the schema with from(schema, table) or the request's schema: the engine does not read a raw Content-Profile header.

A write runs its request lambda up to three times (to read its schema, filter and select(), to tell a dry run, and to send it; an OnlineOnly write, once), so it must configure the same request every time: no side effects, and no clock or counter read inside it.

Seeing and dropping queued writes

From the samples: OfflineTaskRepository.kt · LibraryMappings.kt · TaskRepository.swift · QueueWording.swift

/** Every write in the queue, in the order it will be sent, as rows of the Sync tab. */
override fun waitingChanges(): Flow<List<WaitingChange>> =
    offline.pendingMutations().map { mutations -> mutations.map { it.toWaitingChange() } }


/** Discard: the write is never sent, and its rows show again as the server last sent them. */
override suspend fun discardWaitingChange(id: Long) {
    offline.discardPendingMutation(id)
}


/** A queued write as a row of the Sync tab: what it does, to which row, and since when. */
internal fun PendingMutation.toWaitingChange(): WaitingChange =
    WaitingChange(
        id = id,
        table = table,
        operation = operation.toTaskMutationOperation(),
        createdAt = createdAt,
        rowKey = rowKeys.firstOrNull()?.idOrNull(),
        taskTitle = body?.titleOrNull(), // null for a delete
    )


/** The queue's size as it changes, for [table] alone or for every table when it is null. */
override fun pendingMutationCount(table: String?): Flow<Int> =
    offline.pendingMutationCountFlow(table)

/** The queue's size, read once: the badge has it before the status's first update. */
override suspend fun pendingMutationCountNow(): Int = offline.pendingMutationCount()

/// The writes waiting in the mutation queue, of `table` or of every table, in the order they
/// will be sent, as the queue changes.
func pendingMutations(table: String? = nil) -> PendingMutationSequence {
    client.pendingMutations(table: table)
}

/// How many writes wait for `table`, or for every table, as the queue changes.
func pendingMutationCounts(table: String? = nil) -> PendingMutationCountSequence {
    client.pendingMutationCountSequence(table: table)
}

/// Removes a waiting write without sending it: it is lost, and the rows it changed show again
/// as the server last sent them.
func discardPendingMutation(id: Int64) async throws {
    try await client.discardPendingMutation(id: id)
}

/// A waiting write as a row of the Sync tab: "Added “Buy milk” · 2 min ago", from the
/// write's table, operation, body and creation time.
static func waitingChange(_ mutation: PendingMutation, now: Date) -> String {
    let what = change(mutation.operation, table: mutation.table, body: mutation.body)
    return "\(what) · \(age(of: mutation.createdAt, now: now))"
}

/// How many writes wait, read once: the Sync tab's badge at launch, before the first status.
func pendingMutationCountNow() async throws -> Int {
    try await client.pendingMutationCount()
}

pendingMutations(table), pendingMutationCount(table) and pendingMutationCountFlow(table) show the current user's queued writes, of one table or of all, in the order they will be sent: each a PendingMutation with its id, table, operation, createdAt, the keys of the rows it touches (rowKeys) and its body. discardPendingMutation(id) is destructive, and the way out of a write the server keeps failing, which holds every later write of the user behind it. The queue is never evicted and has no size bound. Sending it is on Replaying the queue; refused writes are on Conflicts and failures.

Behavior matrix

Facts about the engine, one row per condition; the conditions are those of the one-shot read matrix, where warm and cold say whether the queries a write touches are cached.

Condition OnlineOnly QueueIfOffline (default)
Online Sent at once, never behind the queue: Applied. The cached queries are not patched. Recorded in the queue, then sent at once when it heads its user's queue: Applied, and patched into every cached query it touches and every live query it now matches. Rows the server returned are used: an INSERT or UPSERT keeps the columns it wrote beside them; an UPDATE that asked for rows back changes those rows alone, and none when it got none. A write whose filter the cache cannot evaluate marks the lists it may reach stale, and their live readers fetch again. Behind pending writes: Queued, and a pass the engine starts sends it. A server's refusal is thrown and the write leaves the queue. These leave it Queued instead: a 503 or 520; PostgREST's schema cache not knowing the table or a column yet (PGRST205, PGRST204), until the tenth such refusal, which rejects it; and a session token the Auth server will not refresh, since the write was never sent.
Offline, warm Throws the connectivity failure; nothing is queued. Queued, and shown at once in each cached query that admits it; an UPDATE also brings a row cached by another query with the same columns into a list it starts matching. Not queued, a MutationNotQueueableException thrown instead, its cause the connectivity failure: an UPDATE or DELETE with no filter or one the queue cannot record exactly, and a dry run.
Offline, cold As offline, warm. Queued; a reader of a query with nothing cached sees the queued rows it admits.
Network access off (disableNetwork()) Throws NetworkDisabledException; nothing is sent. Queued, and nothing is sent until access is back on.
Session restoring Sent at once, as supabase-kt sends any request meanwhile: on the anon key. Queued for the stored user, online or not, and sent with that user's token once the session settles.

What is waiting and what was refused:

  • pendingMutations, pendingMutationCount and pendingMutationCountFlow show the current user's queued writes, and those with no user recorded that they may take over; never another user's, who cannot discard one by its id either.
  • discardPendingMutation waits for a send of the write in flight, then removes it, unsent: the write is lost.
  • Discarding the write at the head of the queue lets the writes behind it go, with no call from the app.
  • rejectedMutations lists the current user's refused writes only; another user can neither see, retry nor dismiss one by guessing its id.
  • retryRejectedMutation queues the write again, and it is sent.
  • A refusal whose body was too large to keep says so (isRetryable is false), and retrying it throws.
  • A write sent at once that the server refuses throws the server's error, and is not listed: the list is for refusals found on replay.