Skip to content

Commit 7e758fa

Browse files
stevensJourneystevensJourney
authored
fix: iOS Socket issues (#333)
1 parent f2a146b commit 7e758fa

File tree

4 files changed

+168
-73
lines changed

4 files changed

+168
-73
lines changed

‎CHANGELOG.md‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,17 @@
11
# Changelog
22

3+
## 1.11.1
4+
5+
- Fix RSocket connection bugs on iOS (and other platforms using the RSocket sync transport):
6+
- Fix false `connected: true` status when using an invalid token. `ConnectionEstablished` is now
7+
only emitted when the first data frame arrives from the server, matching the HTTP path which
8+
waits for a `200 OK`.
9+
- Fix sync loop terminating permanently when the server rejects the connection with an RSocket
10+
ERROR frame (e.g. invalid JWT). `RSocketError` extends `Throwable` not `Exception`, so it was
11+
not caught by the retry loop.
12+
- Fix sync loop stalling indefinitely after a transport-layer failure (dead socket, network
13+
dropout).
14+
315
## 1.11.0
416

517
- __Breaking__: On tables, the `localOnly`, `insertOnly`, `trackMetadata`, `trackPreviousValues` and

‎common/src/commonMain/kotlin/com/powersync/sync/RSocketSupport.kt‎

Lines changed: 110 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -19,14 +19,16 @@ import io.rsocket.kotlin.transport.RSocketClientTarget
1919
import io.rsocket.kotlin.transport.RSocketConnection
2020
import io.rsocket.kotlin.transport.RSocketTransportApi
2121
import io.rsocket.kotlin.transport.ktor.websocket.internal.KtorWebSocketConnection
22+
import kotlinx.coroutines.CancellationException
2223
import kotlinx.coroutines.Dispatchers
2324
import kotlinx.coroutines.IO
2425
import kotlinx.coroutines.currentCoroutineContext
26+
import kotlinx.coroutines.ensureActive
2527
import kotlinx.coroutines.flow.Flow
2628
import kotlinx.coroutines.flow.emitAll
2729
import kotlinx.coroutines.flow.flow
2830
import kotlinx.coroutines.flow.flowOn
29-
import kotlinx.coroutines.flow.map
31+
import kotlinx.coroutines.flow.transform
3032
import kotlinx.io.readByteArray
3133
import kotlinx.serialization.SerialName
3234
import kotlinx.serialization.Serializable
@@ -50,81 +52,115 @@ internal fun HttpClient.rSocketSyncStream(
5052
credentials: PowerSyncCredentials,
5153
): Flow<PowerSyncControlArguments> =
5254
flow {
53-
val flowContext = currentCoroutineContext()
55+
try {
56+
val flowContext = currentCoroutineContext()
5457

55-
val websocketUri =
56-
URLBuilder(credentials.endpointUri("sync/stream")).apply {
57-
protocol =
58-
when (protocolOrNull) {
59-
URLProtocol.HTTP -> URLProtocol.WS
60-
else -> URLProtocol.WSS
61-
}
62-
}
63-
64-
// Note: We're using a custom connector here because we need to set options for each request
65-
// without creating a new HTTP client each time. The recommended approach would be to add an
66-
// RSocket extension to the HTTP client, but that only allows us to set the SETUP metadata for
67-
// all connections (bad because we need a short-lived token in there).
68-
// https://github.com/rsocket/rsocket-kotlin/issues/311
69-
val target =
70-
object : RSocketClientTarget {
71-
@RSocketTransportApi
72-
override suspend fun connectClient(): RSocketConnection {
73-
val ws =
74-
webSocketSession {
75-
url.takeFrom(websocketUri)
58+
val websocketUri =
59+
URLBuilder(credentials.endpointUri("sync/stream")).apply {
60+
protocol =
61+
when (protocolOrNull) {
62+
URLProtocol.HTTP -> URLProtocol.WS
63+
else -> URLProtocol.WSS
7664
}
77-
return KtorWebSocketConnection(ws)
7865
}
7966

80-
override val coroutineContext: CoroutineContext
81-
get() = flowContext
82-
}
67+
// Note: We're using a custom connector here because we need to set options for each request
68+
// without creating a new HTTP client each time. The recommended approach would be to add an
69+
// RSocket extension to the HTTP client, but that only allows us to set the SETUP metadata for
70+
// all connections (bad because we need a short-lived token in there).
71+
// https://github.com/rsocket/rsocket-kotlin/issues/311
72+
val target =
73+
object : RSocketClientTarget {
74+
@RSocketTransportApi
75+
override suspend fun connectClient(): RSocketConnection {
76+
val ws =
77+
webSocketSession {
78+
url.takeFrom(websocketUri)
79+
}
80+
return KtorWebSocketConnection(ws)
81+
}
8382

84-
val connector =
85-
RSocketConnector {
86-
connectionConfig {
87-
payloadMimeType =
88-
PayloadMimeType(
89-
metadata = "application/json",
90-
data = "application/json",
91-
)
83+
override val coroutineContext: CoroutineContext
84+
get() = flowContext
85+
}
9286

93-
setupPayload {
94-
buildPayload {
95-
data("{}")
96-
metadata(
97-
JsonUtil.json.encodeToString(
98-
ConnectionSetupMetadata(
99-
token = "Bearer ${credentials.token}",
100-
userAgent = userAgent,
101-
),
102-
),
87+
val connector =
88+
RSocketConnector {
89+
connectionConfig {
90+
payloadMimeType =
91+
PayloadMimeType(
92+
metadata = "application/json",
93+
data = "application/json",
10394
)
95+
96+
setupPayload {
97+
buildPayload {
98+
data("{}")
99+
metadata(
100+
JsonUtil.json.encodeToString(
101+
ConnectionSetupMetadata(
102+
token = "Bearer ${credentials.token}",
103+
userAgent = userAgent,
104+
),
105+
),
106+
)
107+
}
104108
}
105-
}
106109

107-
keepAlive = KeepAlive(interval = 20.0.seconds, maxLifetime = 30.0.seconds)
110+
keepAlive = KeepAlive(interval = 20.0.seconds, maxLifetime = 30.0.seconds)
111+
}
108112
}
109-
}
110113

111-
val rSocket = connector.connect(target)
112-
emit(PowerSyncControlArguments.ConnectionEstablished)
113-
val syncStream =
114-
rSocket.requestStream(
115-
buildPayload {
116-
data(JsonUtil.json.encodeToString(req))
117-
metadata(JsonUtil.json.encodeToString(RequestStreamMetadata("/sync/stream")))
118-
},
119-
)
114+
val rSocket = connector.connect(target)
115+
val syncStream =
116+
rSocket.requestStream(
117+
buildPayload {
118+
data(JsonUtil.json.encodeToString(req))
119+
metadata(JsonUtil.json.encodeToString(RequestStreamMetadata("/sync/stream")))
120+
},
121+
)
120122

121-
emitAll(
122-
syncStream
123-
.map {
124-
PowerSyncControlArguments.BinaryLine(it.data.readByteArray())
125-
}.flowOn(Dispatchers.IO),
126-
)
127-
emit(PowerSyncControlArguments.ResponseStreamEnd)
123+
// Emit ConnectionEstablished only when the first frame arrives from the server, mirroring
124+
// the HTTP path which emits it only after receiving a 200 OK. This prevents falsely
125+
// reporting a successful connection when the server rejects the token: the WebSocket
126+
// upgrade (HTTP 101) succeeds at the transport layer, but token validation happens when
127+
// the server processes the stream request and responds.
128+
var connectionEstablishedEmitted = false
129+
emitAll(
130+
syncStream
131+
.transform { payload ->
132+
if (!connectionEstablishedEmitted) {
133+
connectionEstablishedEmitted = true
134+
emit(PowerSyncControlArguments.ConnectionEstablished)
135+
}
136+
emit(PowerSyncControlArguments.BinaryLine(payload.data.readByteArray()))
137+
}.flowOn(Dispatchers.IO),
138+
)
139+
emit(PowerSyncControlArguments.ResponseStreamEnd)
140+
} catch (e: CancellationException) {
141+
// A CancellationException here means the transport layer failed without the server
142+
// sending an RSocket ERROR frame first (e.g. iOS OS detecting a dead socket, airplane
143+
// mode, wifi dropout). rsocket-kotlin cancels its internal connection scope with the
144+
// underlying IOException as the cause, which arrives here as a CancellationException
145+
// rather than an RSocketError. The iOS OS surfaces dead sockets promptly, so this
146+
// path is reached well before any keep-alive timeout fires.
147+
//
148+
// If our own collector coroutine is being cancelled (e.g. disconnect() was called),
149+
// ensureActive() re-throws and the cancellation propagates normally.
150+
// If only the RSocket-internal scope was cancelled, ensureActive() returns normally
151+
// and we wrap as RuntimeException so the enclosing launch{} is treated as *failed*
152+
// rather than *cancelled* — preventing a fetchLinesJob stall.
153+
currentCoroutineContext().ensureActive()
154+
// e.cause is the underlying error (e.g. IOException, RSocketError.ConnectionError)
155+
// that rsocket-kotlin wrapped in a CancellationException via connection.cancel().
156+
// Include it in the message so it surfaces in the streamingSync() error log.
157+
val rootCause = e.cause
158+
val message = "RSocket sync stream was interrupted: ${rootCause?.message ?: e.message}"
159+
if (rootCause?.message?.contains("PSYNC_S21") == true) {
160+
throw RSocketCredentialsExpiredException(message, e)
161+
}
162+
throw RuntimeException(message, e)
163+
}
128164
}
129165

130166
/**
@@ -146,3 +182,13 @@ private class ConnectionSetupMetadata(
146182
private class RequestStreamMetadata(
147183
val path: String,
148184
)
185+
186+
/**
187+
* Thrown from [rSocketSyncStream] when the server closes the RSocket connection with a
188+
* PowerSync authorization error (PSYNC_S21xx) embedded in the transport-level error message.
189+
* Caught by [StreamingSync] to trigger credential invalidation.
190+
*/
191+
internal class RSocketCredentialsExpiredException(
192+
message: String,
193+
cause: Throwable,
194+
) : RuntimeException(message, cause)

‎common/src/commonMain/kotlin/com/powersync/sync/StreamingSync.kt‎

Lines changed: 45 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ import io.ktor.utils.io.ByteReadChannel
3737
import io.ktor.utils.io.readAvailable
3838
import io.ktor.utils.io.readBuffer
3939
import io.ktor.utils.io.readUTF8Line
40+
import io.rsocket.kotlin.RSocketError
4041
import kotlinx.coroutines.CancellationException
4142
import kotlinx.coroutines.CompletableDeferred
4243
import kotlinx.coroutines.CoroutineName
@@ -59,6 +60,7 @@ import kotlinx.coroutines.flow.emitAll
5960
import kotlinx.coroutines.flow.flow
6061
import kotlinx.coroutines.flow.flowOn
6162
import kotlinx.coroutines.flow.map
63+
import kotlinx.coroutines.isActive
6264
import kotlinx.coroutines.launch
6365
import kotlinx.coroutines.withContext
6466
import kotlinx.io.readByteArray
@@ -95,13 +97,16 @@ internal class StreamingSyncClient(
9597

9698
private val httpClient: HttpClient =
9799
when (val config = options.clientConfiguration) {
98-
is SyncClientConfiguration.ExtendedConfig ->
100+
is SyncClientConfiguration.ExtendedConfig -> {
99101
HttpClient {
100102
configureSyncHttpClient(options.userAgent)
101103
config.block(this)
102104
}
105+
}
103106

104-
is SyncClientConfiguration.ExistingClient -> config.client
107+
is SyncClientConfiguration.ExistingClient -> {
108+
config.client
109+
}
105110
}
106111

107112
fun invalidateCredentials() {
@@ -127,11 +132,27 @@ internal class StreamingSyncClient(
127132
invalidCredentials = false
128133
}
129134
result = streamingSyncIteration()
135+
} catch (e: RSocketError) {
136+
// RSocketError extends Throwable directly (not Exception), so it needs its own
137+
// catch block to avoid accidentally catching JVM Errors (OutOfMemoryError, etc.).
138+
if (e is RSocketError.Setup.Rejected) {
139+
// The server rejected the RSocket SETUP frame, most likely due to an invalid
140+
// token. Invalidate credentials so a fresh token is fetched on the next attempt.
141+
connector.invalidateCredentials()
142+
}
143+
logger.e("Error in streamingSync: ${e.message}")
144+
status.update { copy(downloadError = e) }
130145
} catch (e: Exception) {
131146
if (e is CancellationException) {
132147
throw e
133148
}
134149

150+
if (e is RSocketCredentialsExpiredException) {
151+
// Auth error (PSYNC_S21xx) delivered via the RSocket transport-layer failure
152+
// path. Invalidate credentials so a fresh token is fetched on the next attempt.
153+
connector.invalidateCredentials()
154+
}
155+
135156
logger.e("Error in streamingSync: ${e.message}")
136157
status.update { copy(downloadError = e) }
137158
} finally {
@@ -602,17 +623,33 @@ internal class StreamingSyncClient(
602623
state: SyncStreamState,
603624
): SyncStreamState =
604625
when (line) {
605-
is SyncLine.FullCheckpoint -> handleStreamingSyncCheckpoint(line, state)
606-
is SyncLine.CheckpointDiff -> handleStreamingSyncCheckpointDiff(line, state)
607-
is SyncLine.CheckpointComplete -> handleStreamingSyncCheckpointComplete(state)
608-
is SyncLine.CheckpointPartiallyComplete ->
626+
is SyncLine.FullCheckpoint -> {
627+
handleStreamingSyncCheckpoint(line, state)
628+
}
629+
630+
is SyncLine.CheckpointDiff -> {
631+
handleStreamingSyncCheckpointDiff(line, state)
632+
}
633+
634+
is SyncLine.CheckpointComplete -> {
635+
handleStreamingSyncCheckpointComplete(state)
636+
}
637+
638+
is SyncLine.CheckpointPartiallyComplete -> {
609639
handleStreamingSyncCheckpointPartiallyComplete(
610640
line,
611641
state,
612642
)
643+
}
644+
645+
is SyncLine.KeepAlive -> {
646+
handleStreamingKeepAlive(line, state)
647+
}
648+
649+
is SyncLine.SyncDataBucket -> {
650+
handleStreamingSyncData(line, state)
651+
}
613652

614-
is SyncLine.KeepAlive -> handleStreamingKeepAlive(line, state)
615-
is SyncLine.SyncDataBucket -> handleStreamingSyncData(line, state)
616653
SyncLine.UnknownSyncLine -> {
617654
logger.w { "Unhandled instruction $jsonString" }
618655
state

‎gradle.properties‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ development=true
1919
RELEASE_SIGNING_ENABLED=true
2020
# Library config
2121
GROUP=com.powersync
22-
LIBRARY_VERSION=1.11.0
22+
LIBRARY_VERSION=1.11.1
2323
GITHUB_REPO=https://github.com/powersync-ja/powersync-kotlin.git
2424
# POM
2525
POM_URL=https://github.com/powersync-ja/powersync-kotlin/

0 commit comments

Comments
 (0)