Skip to content

Commit 803090d

Browse files
tmmampagent
andauthored
feat(zero-swift): report replayed mutation outcomes (#32)
Add Zero.mutationOutcomes and Zero.addMutationOutcomeListener(_:), which report the server's outcome of each mutation once, with its mutator name and arguments. Unlike a mutation's server result, which settles with .zeroError when the client gives up connecting, they also report the outcome of mutations replayed after the connection comes back. Documented as a Swift-only divergence in JSParity. Amp-Thread-ID: https://ampcode.com/threads/T-c513a037-a257-45e0-b9a9-15c15edfe859 Co-authored-by: Amp <amp@ampcode.com>
1 parent 0f95abb commit 803090d

7 files changed

Lines changed: 177 additions & 19 deletions

File tree

‎.changeset/mutation-outcomes.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
'zero-swift': patch
3+
---
4+
5+
Add `Zero.mutationOutcomes` and `Zero.addMutationOutcomeListener(_:)`, which report the server's outcome of every mutation once, including mutations replayed after a disconnect, whose `server` already settled with `.zeroError`.

‎swift/Sources/Zero/Client/Definitions.swift‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -293,6 +293,21 @@ public enum MutationOutcome: Sendable, Hashable {
293293
}
294294
}
295295

296+
/// The server's outcome of a mutation this client made, reported by
297+
/// ``Zero/mutationOutcomes`` and ``Zero/addMutationOutcomeListener(_:)``.
298+
/// Swift only: the JS client has no counterpart.
299+
public struct MutationServerOutcome: Sendable, Hashable {
300+
/// The mutation's ID, counted per client from 1.
301+
public var mutationID: Int
302+
/// The mutator's name, such as `issue.create`.
303+
public var name: String
304+
/// The mutator's arguments.
305+
public var args: JSONValue?
306+
/// `.success` or `.appError`, with the server's data or message; or
307+
/// `.zeroError` when the server reported a mutation it couldn't process.
308+
public var outcome: MutationOutcome
309+
}
310+
296311
/// What `zero.mutate` returns: the outcome of the optimistic run on the
297312
/// client, and the server's.
298313
@MainActor

‎swift/Sources/Zero/Client/Engine.swift‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -213,6 +213,7 @@ final class Engine {
213213
ackMutations: { [weak self] id in self?.sendIfConnected(.ackMutationResponses(id)) },
214214
onFatalError: { [weak self] error in self?.disconnect(error) })
215215
mutationTracker.setClientID(clientID)
216+
mutationTracker.onOutcome = { [weak self] outcome in self?.onMutationOutcome?(outcome) }
216217
context = QueryContext(
217218
branch: ivmMain,
218219
addCustomQuery: { [weak self] ast, id, ttl, got in
@@ -1017,6 +1018,8 @@ final class Engine {
10171018
/// What `Zero` does when the app must be updated to sync, besides calling
10181019
/// `ZeroOptions.onUpdateNeeded`.
10191020
var onUpdateNeeded: ((UpdateNeededReason) -> Void)?
1021+
/// Hears the server's outcome of each mutation this client made.
1022+
var onMutationOutcome: ((MutationServerOutcome) -> Void)?
10201023

10211024
private func clientStateNotFound() {
10221025
if let onClientStateNotFound {
@@ -1186,7 +1189,7 @@ final class Engine {
11861189
let mutationID = store.nextMutationID(for: clientID)
11871190
let write = KVWrite(store.main)
11881191
let branch = ivmMain.fork()
1189-
mutationTracker.mutationIDAssigned(ephemeralID, mutationID)
1192+
mutationTracker.mutationIDAssigned(ephemeralID, mutationID, name: name, args: args ?? .null)
11901193
let tx = Transaction(
11911194
clientID: clientID, mutationID: Int(mutationID), reason: .optimistic, token: token,
11921195
write: write, branch: branch, schema: schema)

‎swift/Sources/Zero/Client/MutationTracker.swift‎

Lines changed: 41 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,13 @@ final class MutationTracker {
4949
private var clientID: String?
5050
private var largestOutstandingMutationID: Double = 0
5151
private var currentMutationID: Double = 0
52+
/// The mutations sent to the server whose outcome hasn't been reported,
53+
/// by mutation ID. Unlike `outstanding`, they stay when the connection
54+
/// gives up on them: they're sent again on reconnect, and their outcome
55+
/// is reported then.
56+
private var unreported: [Double: (name: String, args: JSONValue?)] = [:]
57+
/// Hears each mutation's outcome from the server once.
58+
var onOutcome: ((MutationServerOutcome) -> Void)?
5259

5360
private let ackMutations: (MutationID) -> Void
5461
private let onFatalError: (ConnectionError) -> Void
@@ -77,8 +84,11 @@ final class MutationTracker {
7784
return (nextEphemeralID, server)
7885
}
7986

80-
func mutationIDAssigned(_ ephemeralID: Int, _ mutationID: Double) {
87+
func mutationIDAssigned(
88+
_ ephemeralID: Int, _ mutationID: Double, name: String, args: JSONValue?
89+
) {
8190
guard let entry = entry(ephemeralID) else { return }
91+
unreported[mutationID] = (name, args)
8292
entry.mutationID = mutationID
8393
ephemeralIDsByMutationID[mutationID] = ephemeralID
8494
largestOutstandingMutationID = max(largestOutstandingMutationID, mutationID)
@@ -87,6 +97,7 @@ final class MutationTracker {
8797
/// Rejects a mutation the client could not run. It was not stored.
8898
func rejectMutation(_ ephemeralID: Int, _ error: any Error) {
8999
guard let entry = entry(ephemeralID) else { return }
100+
if let mid = entry.mutationID { unreported[mid] = nil }
90101
settle(ephemeralID, entry, .failure(ApplicationError.wrapping(error)))
91102
}
92103

@@ -172,6 +183,10 @@ final class MutationTracker {
172183
}
173184
if lastMutationID == currentMutationID { return }
174185
currentMutationID = lastMutationID
186+
// Applied without a result of their own: they succeeded.
187+
for mid in unreported.keys.filter({ $0 <= lastMutationID }).sorted() {
188+
report(mid, .success(nil))
189+
}
175190
while head < order.count {
176191
let id = order[head]
177192
if let entry = outstanding[id] {
@@ -205,29 +220,24 @@ final class MutationTracker {
205220
guard clientID == self.clientID else {
206221
throw ConnectionError.client(.internal, message: "received mutation for the wrong client")
207222
}
208-
guard let ephemeralID = ephemeralIDsByMutationID[mid], let entry = entry(ephemeralID) else {
209-
return
210-
}
211223
let kind = error.error.stringValue ?? ""
212224
if kind == "alreadyProcessed" { return }
225+
let failure: any Error
213226
if kind == "app" {
214-
settle(
215-
ephemeralID, entry,
216-
.failure(
217-
ApplicationError(
218-
error.message?.stringValue ?? "Unknown application error: app",
219-
details: error.details)))
227+
failure = ApplicationError(
228+
error.message?.stringValue ?? "Unknown application error: app", details: error.details)
220229
} else {
221230
let message =
222231
kind == "oooMutation"
223232
? "Server reported an out-of-order mutation"
224233
: "Unknown fallback error with mutation ID \(JSNumber.format(mid)): \(kind)"
225-
settle(
226-
ephemeralID, entry,
227-
.failure(
228-
ConnectionError.server(
229-
.basic(BasicErrorBody(kind: .invalidPush, message: message, origin: .server)))))
234+
failure = ConnectionError.server(
235+
.basic(BasicErrorBody(kind: .invalidPush, message: message, origin: .server)))
236+
}
237+
if let ephemeralID = ephemeralIDsByMutationID[mid], let entry = entry(ephemeralID) {
238+
settle(ephemeralID, entry, .failure(failure))
230239
}
240+
report(mid, .failure(failure))
231241
if kind == "oooMutation" {
232242
onFatalError(
233243
.server(
@@ -242,10 +252,23 @@ final class MutationTracker {
242252
guard clientID == self.clientID else {
243253
throw ConnectionError.client(.internal, message: "received mutation for the wrong client")
244254
}
245-
guard let ephemeralID = ephemeralIDsByMutationID[mid], let entry = entry(ephemeralID) else {
246-
return
255+
if let ephemeralID = ephemeralIDsByMutationID[mid], let entry = entry(ephemeralID) {
256+
settle(ephemeralID, entry, .success(data))
247257
}
248-
settle(ephemeralID, entry, .success(data))
258+
report(mid, .success(data))
259+
}
260+
261+
/// Reports the outcome of mutation `mid`, the first time the server gives
262+
/// one.
263+
private func report(_ mid: Double, _ result: Result<JSONValue?, any Error>) {
264+
guard let m = unreported.removeValue(forKey: mid) else { return }
265+
let outcome: MutationOutcome =
266+
switch result {
267+
case .success(let data): .success(data: data)
268+
case .failure(let error): Engine.outcome(error)
269+
}
270+
onOutcome?(
271+
MutationServerOutcome(mutationID: Int(mid), name: m.name, args: m.args, outcome: outcome))
249272
}
250273

251274
private func settle(_ ephemeralID: Int, _ entry: Entry, _ result: Result<JSONValue?, any Error>) {

‎swift/Sources/Zero/Client/ZeroClient.swift‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,8 @@ public final class Zero<Context: Sendable> {
2121
/// The views and preloads to move to a new client.
2222
private var views: [WeakView] = []
2323
private var preloads: [PreloadRecord] = []
24+
private var outcomeListeners: [(id: Int, listener: (MutationServerOutcome) -> Void)] = []
25+
private var nextOutcomeListenerID = 0
2426
#if canImport(Darwin)
2527
private var lifecycle: AppLifecycle?
2628
#endif
@@ -77,6 +79,45 @@ public final class Zero<Context: Sendable> {
7779
engine.onClientStateNotFound = { [weak self] in self?.replaceClient() }
7880
}
7981
engine.onUpdateNeeded = { [weak self] reason in self?.connection.updateNeeded = reason }
82+
engine.onMutationOutcome = { [weak self] outcome in
83+
for (_, l) in self?.outcomeListeners ?? [] { l(outcome) }
84+
}
85+
}
86+
87+
/// Calls `listener` with the server's outcome of each mutation this
88+
/// client makes from now on, once per mutation, as the server reports it.
89+
///
90+
/// Unlike a mutation's `server` result, it hears the outcome of a
91+
/// mutation sent again after the connection came back: `server` settles
92+
/// with `.zeroError` when the client gives up connecting
93+
/// (`disconnectTimeoutMs`), and the mutation is replayed when it
94+
/// reconnects. A mutation still pending when the app quits is replayed
95+
/// when it next launches, but its outcome isn't reported.
96+
///
97+
/// Use it for a single place that shows rejections, instead
98+
/// of each `server` result, so a rejection isn't shown twice. Swift only:
99+
/// the JS client has no counterpart.
100+
@discardableResult
101+
public func addMutationOutcomeListener(
102+
_ listener: @escaping (MutationServerOutcome) -> Void
103+
) -> () -> Void {
104+
let id = nextOutcomeListenerID
105+
nextOutcomeListenerID += 1
106+
outcomeListeners.append((id, listener))
107+
return { [weak self] in self?.outcomeListeners.removeAll { $0.id == id } }
108+
}
109+
110+
/// The server's outcomes of the mutations this client makes, as
111+
/// ``addMutationOutcomeListener(_:)`` hears them, until the iterating
112+
/// task is cancelled.
113+
public var mutationOutcomes: AsyncStream<MutationServerOutcome> {
114+
let (stream, continuation) = AsyncStream<MutationServerOutcome>.makeStream()
115+
let id = nextOutcomeListenerID
116+
_ = addMutationOutcomeListener { continuation.yield($0) }
117+
continuation.onTermination = { [weak self] _ in
118+
Task { @MainActor in self?.outcomeListeners.removeAll { $0.id == id } }
119+
}
120+
return stream
80121
}
81122

82123
/// Replaces the client, which can't sync any longer, by a new one on the

‎swift/Sources/Zero/Zero.docc/JSParity.md‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -148,6 +148,16 @@ is written or sent.
148148
Freeing one still recurses: in a release build 1,500 levels fit in a
149149
512 KB stack (a macOS test or cooperative thread), but a debug build
150150
overflows one when it frees objects nested more than about 1,000 deep.
151+
- **Mutation outcomes after a disconnect (Swift only).** As in JS, when
152+
the connection drops before the server answers a mutation, its `server`
153+
settles with `.zeroError` and doesn't settle again after the mutation is
154+
replayed. JS has no way to learn the replay's result. Swift adds
155+
``Zero/mutationOutcomes`` and ``Zero/addMutationOutcomeListener(_:)``,
156+
which report the server's outcome of every mutation once, with its
157+
mutator name and arguments, including mutations replayed after their
158+
`server` gave up (but not mutations from before the app last launched).
159+
To show a rejection, use either the stream or `server`,
160+
not both, or the user sees it twice.
151161
- **Storage.** The replica is a SQLite database per user and URL, not
152162
IndexedDB. One process uses it at a time; there are no tabs to share it
153163
with.

‎swift/Tests/ZeroTests/ClientAPITests.swift‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -122,6 +122,67 @@ import ZeroTesting
122122
#expect(try await zero.run(zql.issue).map(\.title) == ["t"])
123123
}
124124

125+
/// The outcomes stream hears each mutation's server outcome once,
126+
/// including those of mutations replayed after `server` gave up on them.
127+
@Test func reportsServerOutcomesOfMutationsReplayedAfterADisconnect() async throws {
128+
let zero = try makeZero()
129+
var outcomes: [MutationServerOutcome] = []
130+
_ = zero.addMutationOutcomeListener { outcomes.append($0) }
131+
func connect() throws -> MockSocket {
132+
scheduler.runDueTimers()
133+
let socket = try #require(sockets.sockets.last)
134+
socket.receive(text: #"["connected",{"wsid":"w"}]"#)
135+
scheduler.runDueTimers()
136+
return socket
137+
}
138+
func poke(_ socket: MockSocket, _ base: String, _ cookie: String, _ part: String) {
139+
socket.receive(text: #"["pokeStart",{"pokeID":"\#(cookie)","baseCookie":\#(base)}]"#)
140+
socket.receive(text: #"["pokePart",{"pokeID":"\#(cookie)",\#(part)}]"#)
141+
socket.receive(text: #"["pokeEnd",{"pokeID":"\#(cookie)","cookie":"\#(cookie)"}]"#)
142+
scheduler.runDueTimers()
143+
}
144+
let cid = zero.clientID
145+
146+
let first = try connect()
147+
let one = zero.mutate(Self.create(CreateIssue(id: "i1", title: "one")))
148+
scheduler.runDueTimers()
149+
poke(first, "null", "01", #""lastMutationIDChanges":{"\#(cid)":1}"#)
150+
#expect(try await one.server.value == .success(data: nil))
151+
152+
// The server goes away with two mutations unconfirmed, and the client
153+
// gives up on it.
154+
let two = zero.mutate(Self.create(CreateIssue(id: "i2", title: "two")))
155+
let three = zero.mutate(Self.create(CreateIssue(id: "i3", title: "three")))
156+
scheduler.runDueTimers()
157+
first.serverClose()
158+
for _ in 0..<70 { scheduler.advance(by: 1_000) }
159+
guard case .zeroError = try await two.server.value else {
160+
Testing.Issue.record("the server result of a mutation the client gave up on")
161+
return
162+
}
163+
guard case .zeroError = try await three.server.value else { return }
164+
165+
// It comes back: the two are replayed, and it rejects one.
166+
scheduler.advance(by: 5_000)
167+
let second = try connect()
168+
poke(
169+
second, #""01""#, "02",
170+
#""lastMutationIDChanges":{"\#(cid)":3},"mutationsPatch":[{"op":"put","mutation":{"id":{"clientID":"\#(cid)","id":2},"result":{"error":"app","message":"taken"}}}]"#
171+
)
172+
#expect(
173+
outcomes == [
174+
MutationServerOutcome(
175+
mutationID: 1, name: "issue.create", args: ["id": "i1", "title": "one"],
176+
outcome: .success(data: nil)),
177+
MutationServerOutcome(
178+
mutationID: 2, name: "issue.create", args: ["id": "i2", "title": "two"],
179+
outcome: .appError(message: "taken", details: nil)),
180+
MutationServerOutcome(
181+
mutationID: 3, name: "issue.create", args: ["id": "i3", "title": "three"],
182+
outcome: .success(data: nil)),
183+
])
184+
}
185+
125186
@Test func anUnregisteredMutatorFailsWithoutWritingOrPushing() async throws {
126187
let zero = try makeZero()
127188
scheduler.runDueTimers()

0 commit comments

Comments
 (0)