diff --git a/CHANGELOG.md b/CHANGELOG.md index 87e0d54435..a4a7aae8c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,26 @@ ## Enhancements + * [phoenix.js] Long polling retries interrupted requests without rejoining + channels while its server session remains alive. Messages are retained until + acknowledged and retried pushes are deduplicated. New `socket.onSuspend` and + `socket.onResume` callbacks report interruptions and recovery without a + channel error. Older servers retain the existing reconnect behavior. During + a rolling deploy, a session that reaches a node with an older server falls + back to that behavior for the rest of the session. + * [Phoenix.Endpoint] Add the `:resumable` WebSocket option. Clients that use the + `resumable: true` option of the `phoenix.js` Socket resume the socket over a new + WebSocket after the connection failed or the heartbeat timed out, without + rejoining channels. Both sides send the messages the other side missed. With + `longPollFallbackMs`, clients continue the socket with long polling when they + cannot resume it over a WebSocket in time. Clients connecting to servers without + the option behave like before. Pass an MFA tuple instead of `true` to decide for + each connection, for example to only let signed-in users resume + * [Phoenix.Socket] Add the optional `c:Phoenix.Socket.resume/3` callback to check + a client that resumes its session after a failed request or WebSocket, for example + to verify that the user is still logged in. Returning `:error` stops the socket, + so the client connects again. Custom transports can implement the new optional + `c:Phoenix.Socket.Transport.handle_resume/2` callback * The longpoll session token is now sent in a `x-phoenix-longpoll-token` header instead of the query string. If you do rolling deploys where both old and new nodes are active at the same time, ensure that you deploy diff --git a/assets/js/phoenix/connection.js b/assets/js/phoenix/connection.js index 3364df6761..6c1ea5d3fc 100644 --- a/assets/js/phoenix/connection.js +++ b/assets/js/phoenix/connection.js @@ -32,9 +32,12 @@ export default class Connection { this.channelsErrored = false // set once the close event was handled, as LongPoll can emit it more than once this.closeHandled = false + this.suspended = false transport.onopen = () => { if(this.isCurrent()){ socket.onConnOpen() } } transport.onerror = error => { if(this.isCurrent()){ socket.onConnError(error) } } transport.onmessage = event => { if(this.isCurrent()){ socket.onConnMessage(event) } } + transport.onsuspend = () => { if(this.isCurrent()){ socket.onConnSuspend(this) } } + transport.onresume = () => { if(this.isCurrent()){ socket.onConnResume(this) } } transport.onclose = event => { if(this.isCurrent() && !this.closeHandled){ this.closeHandled = true diff --git a/assets/js/phoenix/constants.js b/assets/js/phoenix/constants.js index 4a367c12ca..8faf93f8a6 100644 --- a/assets/js/phoenix/constants.js +++ b/assets/js/phoenix/constants.js @@ -6,6 +6,8 @@ export const SOCKET_STATES = {connecting: 0, open: 1, closing: 2, closed: 3} export const MAX_LONGPOLL_BATCH_SIZE = 100; export const DEFAULT_TIMEOUT = 10000 export const WS_CLOSE_NORMAL = 1000 +// the close code of a transport whose session is gone, so the socket connects again +export const CLOSE_SESSION_GONE = 3410 export const CHANNEL_STATES = { closed: "closed", errored: "errored", diff --git a/assets/js/phoenix/index.js b/assets/js/phoenix/index.js index 1c93968260..e6f1fbe278 100644 --- a/assets/js/phoenix/index.js +++ b/assets/js/phoenix/index.js @@ -87,6 +87,36 @@ * socket.onClose( () => console.log("the connection dropped") ) * ``` * + * LongPoll automatically retries interrupted requests while the server keeps + * its session alive. During recovery, channels stay joined and outgoing messages + * are buffered. Use `onSuspend` and `onResume` to show connection status without + * resetting channel state: + * + * ``` + * socket.onSuspend( () => console.log("recovering the connection") ) + * socket.onResume( () => console.log("the connection recovered") ) + * ``` + * + * Recovery requires a server that supports acknowledged long-poll messages. + * With an older server, and for the rest of a session that reached a node + * with an older server during a rolling upgrade, failed requests close the + * socket and channels rejoin. If the session expires, the usual error/close events + * fire and channels rejoin. Push timeouts continue to apply while suspended. + * + * WebSockets resume the same way with the `resumable` option, which requires + * the `:resumable` WebSocket option on the server: + * + * ``` + * let socket = new Socket("/socket", {resumable: true}) + * ``` + * + * When the WebSocket fails, or the heartbeat times out, the socket resumes its + * session over a new WebSocket and both sides send the messages the other side + * missed. With the `longPollFallbackMs` option, the socket continues the session + * with long polling when it cannot resume it over a WebSocket within that time, + * for example when a proxy stops allowing WebSockets. Servers that don't support + * it behave like before. + * * * ## Channel Hooks * @@ -195,6 +225,7 @@ import Channel from "./channel" import LongPoll from "./longpoll" import Presence from "./presence" +import ResumableWebSocket from "./resumable_websocket" import Serializer from "./serializer" import Socket from "./socket" @@ -202,6 +233,7 @@ export { Channel, LongPoll, Presence, + ResumableWebSocket, Serializer, Socket } diff --git a/assets/js/phoenix/longpoll.js b/assets/js/phoenix/longpoll.js index 392fe43e7f..9d2f9721e7 100644 --- a/assets/js/phoenix/longpoll.js +++ b/assets/js/phoenix/longpoll.js @@ -2,11 +2,24 @@ import { SOCKET_STATES, TRANSPORTS, AUTH_TOKEN_PREFIX, - MAX_LONGPOLL_BATCH_SIZE + MAX_LONGPOLL_BATCH_SIZE, + CLOSE_SESSION_GONE } from "./constants" import Ajax from "./ajax" +const BINARY_PREFIX = "base64:" + +// how long we wait before retrying a failed request of a session we can resume +const RETRY_DELAYS = [100, 500, 1000, 2000, 5000] + +let base64ToArrayBuffer = (base64) => { + let binary = atob(base64) + let bytes = new Uint8Array(binary.length) + for(let i = 0; i < binary.length; i++){ bytes[i] = binary.charCodeAt(i) } + return bytes.buffer +} + let arrayBufferToBase64 = (buffer) => { let binary = "" let bytes = new Uint8Array(buffer) @@ -33,10 +46,26 @@ export default class LongPoll { this.currentBatch = null this.currentBatchTimer = null this.batchBuffer = [] + // set once the server tells us that it keeps the messages until we acknowledge them, + // so we can retry a failed request instead of starting a new session + this.resumable = false + this.degraded = false + // the number of the last message we received, and of the last message we sent + this.ack = 0 + this.seq = 0 + this.retries = {GET: 0, POST: 0} + this.retryTimers = new Set() + // set once a retry failed as well, until GET and POST have both recovered + this.interrupted = false + // set while we resume a session that another transport started, see resumeSession + this.resuming = false + this.resumeBuffer = [] this.onopen = function (){ } // noop this.onerror = function (){ } // noop this.onmessage = function (){ } // noop this.onclose = function (){ } // noop + this.onsuspend = function (){ } // noop + this.onresume = function (){ } // noop this.pollEndpoint = this.normalizeEndpoint(endPoint) this.readyState = SOCKET_STATES.connecting // we must wait for the caller to finish setting up our callbacks and timeout properties, @@ -44,6 +73,23 @@ export default class LongPoll { setTimeout(() => this.poll(), 0) } + /** + * @private + * + * Continues a session that another transport started, such as a ResumableWebSocket, + * before the first poll. We resume the session with its token and the number of the + * last message we received, and send the messages the server did not confirm again, + * numbered after the last one it confirmed. The server drops the ones it received. + */ + resumeSession({token, received, confirmed, unconfirmed}){ + this.token = token + this.resumable = true + this.resuming = true + this.ack = received + this.seq = confirmed + this.resumeBuffer = unconfirmed.slice() + } + normalizeEndpoint(endPoint){ return (endPoint .replace("ws://", "http://") @@ -51,8 +97,8 @@ export default class LongPoll { .replace(new RegExp("(.*)\/" + TRANSPORTS.websocket), "$1/" + TRANSPORTS.longpoll)) } - endpointURL(){ - return this.pollEndpoint + endpointURL(params = {}){ + return Ajax.appendParams(this.pollEndpoint, params) } closeAndRetry(error, code, reason, wasClean){ @@ -79,22 +125,64 @@ export default class LongPoll { if(this.authToken){ headers["X-Phoenix-AuthToken"] = this.authToken } - this.ajax("GET", headers, null, () => this.ontimeout(), resp => { + const params = this.resumable ? {ack: this.ack} : {} + // a poll after a failed one resumes the session, so the server checks it again + if(this.resumable && (this.resuming || this.retries.GET > 0)){ params.resume = 1 } + this.ajax("GET", headers, null, params, () => this.retryOr("GET", () => this.poll(), () => this.ontimeout()), resp => { if(resp){ - var {status, token, messages} = resp + var {status, token, messages, last} = resp if(status === 410 && this.token !== null){ // In case we already have a token, this means that our existing session // is gone. We fail so that the client rejoins its channels. - this.closeAndRetry(410, 3410, "session_gone", false) + this.closeAndRetry(410, CLOSE_SESSION_GONE, "session_gone", false) return } this.token = token + if(last !== undefined){ + this.resumable = !this.degraded + } else if(this.resuming && (status === 200 || status === 204)){ + // TODO: Remove in Phoenix 1.10, together with degrade() + // + // a node with an older server cannot resume a session of another transport + this.closeAndRetry(410, CLOSE_SESSION_GONE, "session_gone", false) + return + } else if(this.resumable && (status === 200 || status === 204)){ + // TODO: Remove in Phoenix 1.10, when no cluster mixes 1.8 and newer nodes. + // + // The poll went through a node with an older server, as during a rolling + // deploy, which forwards it without our ack. The session then dropped the + // messages we did not acknowledge yet, like it does for older clients. + // We got them unless this poll is a retry, as they were in the response we lost. + if(!this.degrade()){ return } + } } else { status = 0 } - + if(this.resuming && (status === 200 || status === 204)){ + // we resumed the session, so we are open and send what we kept + this.readyState = SOCKET_STATES.open + // keep callback pushes buffered after the messages we already kept + this.onopen({}) + if(!this.isActive()){ return } + this.resuming = false + const buffered = this.resumeBuffer + this.resumeBuffer = [] + buffered.forEach(body => this.send(body)) + } switch(status){ case 200: + if(last !== undefined){ + // The server numbers its messages and sends all we did not acknowledge, + // so we drop the ones we already received, as when a late poll returns. + const first = last - messages.length + 1 + if(first > this.ack + 1){ + // we cannot deliver the messages, as we lost the ones before them + this.closeAndRetry(0, 1011, "lost messages", false) + return + } + messages = messages.slice(Math.max(0, this.ack + 1 - first)) + this.ack = Math.max(this.ack, last) + } messages.forEach(msg => { // Tasks are what things like event handlers, setTimeout callbacks, // promise resolves and more are run within. @@ -117,15 +205,20 @@ export default class LongPoll { // // We might stop before that, e.g. when a POST times out or a message // handler disconnects, which cancels the remaining messages. - const timer = setTimeout(() => { - this.messageTimers.delete(timer) - this.onmessage({data: msg}) - }, 0) - this.messageTimers.add(timer) + // the server sends binary messages in base64, such as the ones a session + // has when it started over a WebSocket + const data = msg.startsWith(BINARY_PREFIX) ? base64ToArrayBuffer(msg.slice(BINARY_PREFIX.length)) : msg + this.scheduleMessage(() => this.onmessage({data})) + }) + // Restore only after delivering the messages, so hooks see their updates. + // Closing from a message callback cancels this task as well. + this.scheduleMessage(() => { + this.succeeded("GET") + this.poll() }) - this.poll() break case 204: + this.succeeded("GET") this.poll() break case 410: @@ -138,6 +231,9 @@ export default class LongPoll { this.close(1008, "forbidden", false) break case 0: + // the connection failed, but the server still has our session and messages + this.retryOr("GET", () => this.poll(), () => this.closeAndRetry(500, 1011, "internal server error", false)) + break case 500: this.closeAndRetry(500, 1011, "internal server error", false) break @@ -146,44 +242,134 @@ export default class LongPoll { }) } + scheduleMessage(callback){ + const timer = setTimeout(() => { + this.messageTimers.delete(timer) + callback() + }, 0) + this.messageTimers.add(timer) + } + // we collect all pushes within the current event loop by // setTimeout 0, which optimizes back-to-back procedural // pushes against an empty buffer send(body){ + // we send once we resumed the session, numbered after the messages we kept + if(this.resuming){ + this.resumeBuffer.push(body) + return + } if(typeof(body) !== "string"){ body = arrayBufferToBase64(body) } + // messages are numbered, so the server drops the ones it received when we send + // them again. A batch is sent with the seq of its first message. + this.seq++ if(this.currentBatch){ this.currentBatch.push(body) } else if(this.awaitingBatchAck){ + if(this.batchBuffer.length === 0){ this.batchBufferSeq = this.seq } this.batchBuffer.push(body) } else { this.currentBatch = [body] + const seq = this.seq this.currentBatchTimer = setTimeout(() => { - this.batchSend(this.currentBatch) + this.batchSend(this.currentBatch, seq) this.currentBatch = null }, 0) } } - batchSend(messages, offset = 0){ + batchSend(messages, seq, offset = 0){ + if(!this.isActive()){ return } this.awaitingBatchAck = true const next = offset + MAX_LONGPOLL_BATCH_SIZE const batch = messages.slice(offset, next) - this.ajax("POST", {"Content-Type": "application/x-ndjson"}, batch.join("\n"), () => this.ontimeout(), resp => { - if(!resp || resp.status !== 200){ + const retry = () => this.batchSend(messages, seq, offset) + // servers that don't number their messages don't know about seq either + const params = this.resumable ? {seq: seq + offset} : {} + this.ajax("POST", {"Content-Type": "application/x-ndjson"}, batch.join("\n"), params, () => this.retryOr("POST", retry, () => this.ontimeout()), resp => { + if(!resp){ + this.retryOr("POST", retry, () => this.closeAndRetry(null, 1011, "internal server error", false)) + } else if(resp.status !== 200){ // this calls stop which already clears awaitingBatchAck this.closeAndRetry(resp && resp.status, 1011, "internal server error", false) - } else if(next < messages.length){ - this.batchSend(messages, next) - } else if(this.batchBuffer.length > 0){ - this.batchSend(this.batchBuffer) - this.batchBuffer = [] } else { - this.awaitingBatchAck = false + // a node with an older server forwards the request without our seq, so the + // session does not drop the messages we send again + // + // TODO: Remove in Phoenix 1.10, together with degrade() + if(this.resumable && resp.seq === undefined && !this.degrade()){ return } + this.succeeded("POST") + if(next < messages.length){ + this.batchSend(messages, seq, next) + } else if(this.batchBuffer.length > 0){ + this.batchSend(this.batchBuffer, this.batchBufferSeq) + this.batchBuffer = [] + } else { + this.awaitingBatchAck = false + } } }) } + /** + * @private + * + * Stops resuming for the rest of the session. A node with an older server, as + * during a rolling deploy, does not forward our ack and seq. Retrying through it + * could lose or duplicate messages, so we fail like we do with an older server. + * + * TODO: Remove in Phoenix 1.10, when no cluster mixes 1.8 and newer nodes. + */ + degrade(){ + this.degraded = true + this.resumable = false + // A request already being retried may have been processed. Neither replaying + // it without deduplication nor accepting a gap in received messages is safe. + if(this.retries.GET > 0 || this.retries.POST > 0){ + this.closeAndRetry(0, 1011, "session no longer resumable", false) + return false + } + return true + } + + /** + * @private + * + * Retries a failed request with backoff when the server keeps our session, + * otherwise fails. + */ + retryOr(method, retry, fail){ + if(!this.resumable){ return fail() } + const delay = RETRY_DELAYS[Math.min(this.retries[method], RETRY_DELAYS.length - 1)] + this.retries[method]++ + // a single failed request is retried quickly, so we only report the interruption + // once its retry failed as well + if(this.retries[method] > 1 && !this.interrupted){ + this.interrupted = true + this.onsuspend() + } + if(!this.isActive()){ return } + const timer = setTimeout(() => { + this.retryTimers.delete(timer) + if(!this.resumable){ return fail() } + retry() + }, delay) + this.retryTimers.add(timer) + } + + /** + * @private + */ + succeeded(method){ + this.retries[method] = 0 + // A successful poll must not hide an outstanding failed POST, or vice versa. + if(this.interrupted && this.retries.GET === 0 && this.retries.POST === 0){ + this.interrupted = false + this.onresume() + } + } + close(code, reason, wasClean){ this.stop(SOCKET_STATES.closed) this.emitClose(code, reason, wasClean) @@ -201,6 +387,9 @@ export default class LongPoll { for(let {req} of reqs){ req && req.abort() } for(let timer of this.messageTimers){ clearTimeout(timer) } this.messageTimers.clear() + for(let timer of this.retryTimers){ clearTimeout(timer) } + this.retryTimers.clear() + this.interrupted = false this.batchBuffer = [] this.awaitingBatchAck = false clearTimeout(this.currentBatchTimer) @@ -221,7 +410,7 @@ export default class LongPoll { } } - ajax(method, headers, body, onCallerTimeout, callback){ + ajax(method, headers, body, params, onCallerTimeout, callback){ // the request is ours until it completes or we stop, and the callbacks of a request // that is not ours anymore are ignored. It is ours before it starts, as a request can // complete while it starts. @@ -233,7 +422,7 @@ export default class LongPoll { if(this.token !== null){ headers = Object.assign({}, headers, {"X-Phoenix-Longpoll-Token": this.token}) } - request.req = Ajax.request(method, this.endpointURL(), headers, body, this.timeout, ontimeout, resp => { + request.req = Ajax.request(method, this.endpointURL(params), headers, body, this.timeout, ontimeout, resp => { if(this.reqs.delete(request)){ callback(resp) } }) } diff --git a/assets/js/phoenix/resumable_websocket.js b/assets/js/phoenix/resumable_websocket.js new file mode 100644 index 0000000000..eef644be4e --- /dev/null +++ b/assets/js/phoenix/resumable_websocket.js @@ -0,0 +1,396 @@ +import { + CLOSE_SESSION_GONE, + global, + SOCKET_STATES +} from "./constants" + +import Ajax from "./ajax" +import LongPoll from "./longpoll" + +// how long we wait before resuming the session over a new WebSocket +const RETRY_DELAYS = [100, 500, 1000, 2000, 5000] +// how often we tell the server about the messages we received +const ACK_INTERVAL_MS = 1000 +const DEFAULT_RESUME_TIMEOUT_MS = 10000 +// control messages are V2 messages without refs on the "phoenix" topic +const CONTROL_PREFIX = "[null,null,\"phoenix\",\"phx_" + +let control = (event, payload) => JSON.stringify([null, null, "phoenix", event, payload]) + +/** + * A WebSocket transport that resumes its session over a new WebSocket when the + * connection fails, without rejoining channels. + * + * The server keeps the socket and its channels for a while after the WebSocket + * failed, see the `:resumable` option of `Phoenix.Endpoint.socket/3`. Both sides + * number their messages and keep them until the other side confirmed them, so + * they send the missed ones again after resuming. + * + * With `longPollFallbackMs`, which the socket sets from its option of the same name, + * we continue the session with long polling when resuming it over a WebSocket did + * not succeed within that time, and keep long polling for the rest of the session. + * + * Servers that don't keep the socket for us make it behave like a WebSocket. + */ +export default class ResumableWebSocket { + constructor(endPoint, protocols){ + this.endPoint = endPoint + this.protocols = protocols + this.binaryTypeValue = "arraybuffer" + this.timeout = DEFAULT_RESUME_TIMEOUT_MS + this.readyState = SOCKET_STATES.connecting + // the token of the session, once the server told us that it keeps the socket for us + this.token = null + // set when the server does not keep the socket for us + this.unsupported = false + // the number of messages we received + this.received = 0 + // the messages the server did not confirm yet, and the number of the last one it confirmed + this.unconfirmed = [] + this.confirmed = 0 + this.ws = null + // set while the WebSocket carries our messages, once it opened or the session resumed over it + this.ready = false + // set while we wait for the server to resume the session over the WebSocket + this.resuming = false + this.retries = 0 + this.retryTimer = null + this.resumeTimer = null + this.ackTimer = null + // set once a retry failed as well, until we resumed the session + this.interrupted = false + // how long we try to resume the session over a WebSocket before we long poll, if at all + this.longPollFallbackMs = null + this.fallbackTimer = null + // the LongPoll that continues the session once we fell back to it + this.longpoll = null + this.onopen = function (){ } // noop + this.onerror = function (){ } // noop + this.onmessage = function (){ } // noop + this.onclose = function (){ } // noop + this.onsuspend = function (){ } // noop + this.onresume = function (){ } // noop + this.connect(false) + } + + get binaryType(){ return this.binaryTypeValue } + + set binaryType(binaryType){ + this.binaryTypeValue = binaryType + if(this.ws){ this.ws.binaryType = binaryType } + } + + get bufferedAmount(){ return this.ws ? this.ws.bufferedAmount : 0 } + + send(data){ + if(this.readyState !== SOCKET_STATES.open){ return } + if(this.longpoll){ + this.longpoll.send(data) + return + } + // we number the messages implicitly, in the order we send them + if(!this.unsupported){ this.unconfirmed.push(data) } + if(this.ready){ this.ws.send(data) } + } + + close(code, reason){ + if(this.readyState === SOCKET_STATES.closed){ return } + // the server stops the session, as we did not lose the connection + if(this.ready && this.token){ this.ws.send(control("phx_close", {})) } + if(this.longpoll){ + const longpoll = this.longpoll + this.longpoll = null + longpoll.onclose = function (){ } // noop + longpoll.close(code, reason) + } + this.stop(code, reason) + this.onclose({code: code || 1005, reason: reason || "", wasClean: true}) + } + + /** + * @private + * + * Resumes the session over a new WebSocket when the heartbeat timed out, as the + * current one is presumed dead. Returns whether we resume the session. + */ + resumeAfterHeartbeatTimeout(){ + if(this.readyState !== SOCKET_STATES.open || !this.token){ return false } + // long polling detects failed requests itself + if(this.longpoll){ return true } + if(this.ws && !this.resuming){ + this.dropWebSocket() + this.retry() + } + return true + } + + /** + * @private + */ + connect(resume){ + const WebSocket = ResumableWebSocket.WebSocket || global.WebSocket + const params = resume ? {resumable: 1, resume: 1} : {resumable: 1} + const ws = new WebSocket(Ajax.appendParams(this.endPoint, params), this.protocols) + ws.binaryType = this.binaryTypeValue + this.ws = ws + this.ready = false + this.resuming = resume + if(resume){ + // a server that does not answer is as good as a failed connection + this.resumeTimer = setTimeout(() => { + this.dropWebSocket() + this.retry() + }, this.timeout) + } + ws.onopen = () => { + if(this.ws !== ws){ return } + if(resume){ + ws.send(control("phx_resume", {token: this.token, ack: this.received})) + } else { + this.ready = true + this.readyState = SOCKET_STATES.open + this.onopen() + } + } + ws.onerror = error => { + // failed attempts to resume the session are no errors of the socket + if(this.ws === ws && !this.token){ this.onerror(error) } + } + ws.onmessage = event => { if(this.ws === ws){ this.handleMessage(event) } } + ws.onclose = event => { if(this.ws === ws){ this.handleClose(event) } } + } + + /** + * @private + */ + handleMessage(event){ + const {data} = event + if(typeof(data) === "string" && data.startsWith(CONTROL_PREFIX)){ + const [, , , name, payload] = JSON.parse(data) + this.handleControl(name, payload) + return + } + if(this.resuming){ + // an older server or a socket that is not resumable treats our resume + // as a new connection + this.end() + return + } + // the server starts a session with its token, which is its first message + if(this.token === null && !this.unsupported){ + this.unsupported = true + this.unconfirmed = [] + } + this.received++ + this.scheduleAck() + this.onmessage(event) + } + + /** + * @private + */ + handleControl(name, payload){ + switch(name){ + case "phx_session": + this.token = payload.token + break + case "phx_resumed": + this.resumed(payload.seq) + break + case "phx_ack": + this.confirm(payload.seq) + break + case "phx_gone": + // the session stopped with a close code, or we cannot resume it + this.end(payload.code) + break + default: + // an older server replies to our resume like to an unknown message + if(this.resuming){ this.end() } + } + } + + /** + * @private + */ + resumed(seq){ + if(!this.resuming){ return } + if(seq < this.confirmed || seq > this.confirmed + this.unconfirmed.length){ + // the server cannot have received messages we did not send + this.end() + return + } + clearTimeout(this.resumeTimer) + this.confirm(seq) + this.resuming = false + this.ready = true + // we send the messages the server did not receive again + this.unconfirmed.forEach(data => this.ws.send(data)) + this.retries = 0 + this.recovered() + } + + /** + * @private + */ + interrupt(){ + if(!this.interrupted){ + this.interrupted = true + this.onsuspend() + } + } + + /** + * @private + */ + recovered(){ + clearTimeout(this.fallbackTimer) + this.fallbackTimer = null + if(this.interrupted){ + this.interrupted = false + this.onresume() + } + } + + /** + * @private + */ + confirm(seq){ + if(seq > this.confirmed){ + this.unconfirmed.splice(0, seq - this.confirmed) + this.confirmed = seq + } + } + + /** + * @private + */ + scheduleAck(){ + if(this.ackTimer || !this.token){ return } + this.ackTimer = setTimeout(() => { + this.ackTimer = null + if(this.ready){ this.ws.send(control("phx_ack", {ack: this.received})) } + }, ACK_INTERVAL_MS) + } + + /** + * @private + */ + handleClose(event){ + this.ws = null + this.ready = false + clearTimeout(this.ackTimer) + clearTimeout(this.resumeTimer) + this.ackTimer = null + if(this.readyState === SOCKET_STATES.open && this.token){ + this.retry() + } else { + this.stop() + this.onclose(event) + } + } + + /** + * @private + * + * Resumes the session over a new WebSocket with backoff. + */ + retry(){ + this.resuming = false + const delay = RETRY_DELAYS[Math.min(this.retries, RETRY_DELAYS.length - 1)] + this.retries++ + // a single failed WebSocket is resumed quickly, so we only report the + // interruption once resuming failed as well + if(this.retries > 1){ this.interrupt() } + if(this.readyState !== SOCKET_STATES.open){ return } + if(this.longPollFallbackMs && !this.fallbackTimer){ + this.fallbackTimer = setTimeout(() => this.fallBack(), this.longPollFallbackMs) + } + this.retryTimer = setTimeout(() => { + this.retryTimer = null + this.connect(true) + }, delay) + } + + /** + * @private + * + * Continues the session with long polling, as resuming it over a WebSocket did not + * succeed in time. The LongPoll sends the messages the server did not confirm. + */ + fallBack(){ + this.fallbackTimer = null + if(this.readyState !== SOCKET_STATES.open){ return } + clearTimeout(this.retryTimer) + this.retryTimer = null + this.dropWebSocket() + this.resuming = false + const longpoll = new LongPoll(this.endPoint, this.protocols) + longpoll.timeout = this.timeout + longpoll.resumeSession({ + token: this.token, + received: this.received, + confirmed: this.confirmed, + unconfirmed: this.unconfirmed + }) + this.unconfirmed = [] + this.longpoll = longpoll + const current = () => this.longpoll === longpoll + longpoll.onopen = () => { if(current()){ this.recovered() } } + longpoll.onmessage = event => { if(current()){ this.onmessage(event) } } + longpoll.onerror = error => { if(current()){ this.onerror(error) } } + longpoll.onsuspend = () => { if(current()){ this.interrupt() } } + longpoll.onresume = () => { if(current()){ this.recovered() } } + longpoll.onclose = event => { + if(!current()){ return } + // the session is gone, so the socket connects again + this.longpoll = null + this.stop() + this.onclose(event) + } + } + + /** + * @private + * + * Closes the transport when the session is gone, so the socket connects again. + */ + end(code){ + this.stop() + if(code){ + this.onclose({code, reason: "", wasClean: true}) + } else { + this.onclose({code: CLOSE_SESSION_GONE, reason: "session_gone", wasClean: false}) + } + } + + /** + * @private + */ + dropWebSocket(code, reason){ + const ws = this.ws + this.ws = null + this.ready = false + clearTimeout(this.resumeTimer) + if(!ws){ return } + ws.onopen = ws.onerror = ws.onmessage = ws.onclose = function (){ } // noop + if(ws.readyState === SOCKET_STATES.connecting || ws.readyState === SOCKET_STATES.open){ + if(code){ ws.close(code, reason || "") } else { ws.close() } + } + } + + /** + * @private + */ + stop(code, reason){ + this.readyState = SOCKET_STATES.closed + clearTimeout(this.retryTimer) + clearTimeout(this.ackTimer) + clearTimeout(this.fallbackTimer) + this.retryTimer = null + this.ackTimer = null + this.fallbackTimer = null + this.interrupted = false + this.dropWebSocket(code, reason) + } +} diff --git a/assets/js/phoenix/socket.js b/assets/js/phoenix/socket.js index 65b9356611..172f219264 100644 --- a/assets/js/phoenix/socket.js +++ b/assets/js/phoenix/socket.js @@ -17,6 +17,7 @@ import { import Ajax from "./ajax" import Channel from "./channel" import LongPoll from "./longpoll" +import ResumableWebSocket from "./resumable_websocket" import Serializer from "./serializer" import Timer from "./timer" import Connection from "./connection" @@ -37,6 +38,12 @@ import Connection from "./connection" * @param {number} [opts.longPollFallbackMs] - The millisecond time to attempt the primary transport * before falling back to the LongPoll transport. Disabled by default. * + * @param {boolean} [opts.resumable] - When true, uses the ResumableWebSocket transport, which + * resumes the socket over a new WebSocket when the connection fails, without rejoining channels. + * Requires the `:resumable` WebSocket option on the server. Disabled by default. + * With `longPollFallbackMs`, the socket continues with long polling when it cannot resume + * over a WebSocket within that time. + * * @param {boolean} [opts.debug] - When true, enables debug logging. Default false. * * @param {Function} [opts.encode] - The function to encode outgoing messages. @@ -120,14 +127,14 @@ import Connection from "./connection" */ export default class Socket { constructor(endPoint, opts = {}){ - this.stateChangeCallbacks = {open: [], close: [], error: [], message: []} + this.stateChangeCallbacks = {open: [], close: [], error: [], message: [], suspend: [], resume: []} this.channels = [] this.sendBuffer = [] this.ref = 0 this.fallbackRef = null this.fallbackErrorRef = null this.timeout = opts.timeout || DEFAULT_TIMEOUT - this.transport = opts.transport || global.WebSocket || LongPoll + this.transport = opts.transport || (global.WebSocket && opts.resumable ? ResumableWebSocket : global.WebSocket) || LongPoll this.primaryPassedHealthCheck = false this.longPollFallbackMs = opts.longPollFallbackMs this.fallbackTimer = null @@ -394,6 +401,28 @@ export default class Socket { return ref } + /** + * Registers a callback when the transport is interrupted but keeps its session + * and retries. Channels remain joined and pushes are buffered by the transport. + * Currently supported by LongPoll. A lost session instead emits error/close. + * @param {Function} callback + */ + onSuspend(callback){ + let ref = this.makeRef() + this.stateChangeCallbacks.suspend.push([ref, callback]) + return ref + } + + /** + * Registers a callback when a suspended transport recovers the same session. + * @param {Function} callback + */ + onResume(callback){ + let ref = this.makeRef() + this.stateChangeCallbacks.resume.push([ref, callback]) + return ref + } + /** * Registers callbacks for connection message events * @param {Function} callback @@ -433,6 +462,7 @@ export default class Socket { // is native to browsers and does not need special handling. switch(transport){ case LongPoll: return "LongPoll" + case ResumableWebSocket: return "ResumableWebSocket" default: return transport.name } } @@ -456,6 +486,8 @@ export default class Socket { const transport = new this.transport(this.endPointURL(), protocols) transport.binaryType = this.binaryType transport.timeout = this.longpollerTimeout + // a transport that keeps its session falls back to long polling to continue it + transport.longPollFallbackMs = this.longPollFallbackMs this.connection = new Connection(this, transport) // connect can replace an ended connection that is still open while its buffer drains, but the // channels joined over it need to rejoin on the new connection instead of waiting for it to close @@ -559,6 +591,13 @@ export default class Socket { if(this.pendingHeartbeatRef){ const connection = this.connection this.pendingHeartbeatRef = null + // a transport that keeps its session resumes it over a new connection instead + const transport = connection && connection.transport + if(transport && transport.resumeAfterHeartbeatTimeout && transport.resumeAfterHeartbeatTimeout()){ + if(this.hasLogger()){ this.log("transport", "heartbeat timeout. Resuming the session") } + this.resetHeartbeat() + return + } if(this.hasLogger()){ this.log("transport", "heartbeat timeout. Attempting to re-establish connection") } this.closeWasClean = false // the connection is presumed dead, so we close it right away instead of waiting for its @@ -692,6 +731,28 @@ export default class Socket { this.triggerChanError("connection_error", connection) } + /** @private */ + onConnSuspend(connection){ + if(!this.stillUses(connection) || !connection.isOpen() || connection.suspended){ return } + connection.suspended = true + if(this.hasLogger()) this.log("transport", "suspended") + for(let [, callback] of this.stateChangeCallbacks.suspend.slice()){ + if(!this.stillUses(connection) || !connection.isOpen()){ break } + callback() + } + } + + /** @private */ + onConnResume(connection){ + if(!this.stillUses(connection) || !connection.isOpen() || !connection.suspended){ return } + connection.suspended = false + if(this.hasLogger()) this.log("transport", "resumed") + for(let [, callback] of this.stateChangeCallbacks.resume.slice()){ + if(!this.stillUses(connection) || !connection.isOpen()){ break } + callback() + } + } + /** * @private * diff --git a/assets/test/longpoll_test.js b/assets/test/longpoll_test.js index 98f7d009d6..fc1b049af6 100644 --- a/assets/test/longpoll_test.js +++ b/assets/test/longpoll_test.js @@ -656,6 +656,372 @@ describe("LongPoll", () => { }) }) +describe("LongPoll resuming", () => { + let longpoll, events + + const requests = method => Ajax.request.mock.calls.filter(([m]) => m === method) + const last = method => requests(method).at(-1) + const params = call => Object.fromEntries(new URL(call[1]).searchParams) + const respond = (call, resp) => call[6](resp) + + // the server keeps the messages until we acknowledge them when it numbers them + const open = (resp = {status: 410, token: "token", messages: [], last: 0}) => { + longpoll = new LongPoll("http://localhost/socket/longpoll") + events = [] + longpoll.onopen = () => events.push("open") + longpoll.onmessage = ({data}) => events.push(data) + longpoll.onerror = () => events.push("error") + longpoll.onclose = () => events.push("close") + longpoll.onsuspend = () => events.push("suspend") + longpoll.onresume = () => events.push("resume") + jest.advanceTimersByTime(0) + respond(last("GET"), resp) + } + + beforeEach(() => { + jest.useFakeTimers() + jest.spyOn(Ajax, "request").mockImplementation(() => ({abort: jest.fn()})) + }) + + afterEach(() => { + longpoll.close() + jest.useRealTimers() + jest.restoreAllMocks() + }) + + it("acknowledges the last message it received with the next poll", () => { + open() + expect(params(last("GET")).ack).toBe("0") + + respond(last("GET"), {status: 200, token: "token", messages: ["a"], last: 1}) + jest.advanceTimersByTime(0) + expect(params(last("GET")).ack).toBe("1") + + // a poll without new messages keeps the ack + respond(last("GET"), {status: 204, token: "token", last: 1}) + expect(params(last("GET")).ack).toBe("1") + }) + + // TODO: Remove in Phoenix 1.10, together with degrade() + it("does not acknowledge when the server does not number its messages", () => { + open({status: 410, token: "token", messages: []}) + expect(params(last("GET")).ack).toBeUndefined() + }) + + it.each([ + ["gets no response", call => respond(call, null)], + ["times out", call => call[5]()] + ])("polls again with the same ack when a poll %s", (_, fail) => { + open() + respond(last("GET"), {status: 200, token: "token", messages: ["a"], last: 1}) + jest.advanceTimersByTime(0) + const polls = requests("GET").length + + fail(last("GET")) + jest.advanceTimersByTime(99) + expect(requests("GET")).toHaveLength(polls) + jest.advanceTimersByTime(1) + + expect(requests("GET")).toHaveLength(polls + 1) + expect(params(last("GET")).ack).toBe("1") + expect(events).not.toContain("error") + expect(events).not.toContain("close") + }) + + it("backs off while requests keep failing", () => { + open() + const delays = [] + for(let i = 0; i < 6; i++){ + const polls = requests("GET").length + respond(last("GET"), null) + let waited = 0 + while(requests("GET").length === polls){ + jest.advanceTimersByTime(100) + waited += 100 + } + delays.push(waited) + } + + expect(delays).toEqual([100, 500, 1000, 2000, 5000, 5000]) + }) + + it("reports suspension once a retry failed as well, and resumption once requests succeed", () => { + open() + respond(last("GET"), null) + jest.advanceTimersByTime(100) + expect(events).toEqual(["open"]) + + respond(last("GET"), null) + jest.advanceTimersByTime(500) + respond(last("GET"), null) + jest.advanceTimersByTime(1000) + expect(events).toEqual(["open", "suspend"]) + + respond(last("GET"), {status: 204, token: "token", last: 0}) + expect(events).toEqual(["open", "suspend", "resume"]) + + // a single failure after that is not an interruption yet + respond(last("GET"), null) + jest.advanceTimersByTime(100) + expect(events).toEqual(["open", "suspend", "resume"]) + }) + + it("waits for both GET and POST to recover", () => { + open() + longpoll.send("a") + jest.advanceTimersByTime(0) + respond(last("GET"), null) + respond(last("POST"), null) + jest.advanceTimersByTime(100) + respond(last("GET"), null) + respond(last("POST"), null) + jest.advanceTimersByTime(500) + expect(events).toEqual(["open", "suspend"]) + + respond(last("GET"), {status: 204, token: "token", last: 0}) + expect(events).toEqual(["open", "suspend"]) + respond(last("POST"), {status: 200, seq: 1}) + expect(events).toEqual(["open", "suspend", "resume"]) + }) + + it("delivers the recovered messages before reporting resumption", () => { + open() + respond(last("GET"), null) + jest.advanceTimersByTime(100) + respond(last("GET"), null) + jest.advanceTimersByTime(500) + respond(last("GET"), {status: 200, token: "token", messages: ["a", "b"], last: 2}) + expect(events).toEqual(["open", "suspend"]) + jest.advanceTimersByTime(0) + expect(events).toEqual(["open", "suspend", "a", "b", "resume"]) + expect(params(last("GET")).ack).toBe("2") + }) + + it("does not deliver already received messages again", () => { + open() + respond(last("GET"), {status: 200, token: "token", messages: ["a"], last: 1}) + jest.advanceTimersByTime(0) + respond(last("GET"), {status: 200, token: "token", messages: ["a"], last: 1}) + jest.advanceTimersByTime(0) + expect(events).toEqual(["open", "a"]) + + // a late poll returns the messages after an older ack, with new ones + respond(last("GET"), {status: 200, token: "token", messages: ["a", "b"], last: 2}) + jest.advanceTimersByTime(0) + expect(events).toEqual(["open", "a", "b"]) + expect(params(last("GET")).ack).toBe("2") + }) + + it("closes when messages are missing before the ones it receives", () => { + open() + respond(last("GET"), {status: 200, token: "token", messages: ["c"], last: 3}) + jest.advanceTimersByTime(0) + expect(events).toEqual(["open", "error", "close"]) + }) + + it("resumes the session with the polls after a failed one", () => { + open() + expect(params(last("GET")).resume).toBeUndefined() + respond(last("GET"), null) + jest.advanceTimersByTime(100) + expect(params(last("GET")).resume).toBe("1") + respond(last("GET"), null) + jest.advanceTimersByTime(500) + expect(params(last("GET")).resume).toBe("1") + + respond(last("GET"), {status: 204, token: "token", last: 0}) + expect(params(last("GET")).resume).toBeUndefined() + }) + + // TODO: Remove in Phoenix 1.10, together with degrade() + it("does not resume the session when the server does not number its messages", () => { + open({status: 410, token: "token", messages: []}) + expect(params(last("GET")).resume).toBeUndefined() + }) + + it("does not retry when a suspension callback closes the transport", () => { + open() + longpoll.onsuspend = () => longpoll.close() + respond(last("GET"), null) + jest.advanceTimersByTime(100) + respond(last("GET"), null) + const polls = requests("GET").length + jest.advanceTimersByTime(10_000) + expect(requests("GET")).toHaveLength(polls) + expect(longpoll.retryTimers.size).toBe(0) + }) + + it.each([403, 500])("does not report resumption before failing with status %s", status => { + open() + respond(last("GET"), null) + jest.advanceTimersByTime(100) + respond(last("GET"), null) + jest.advanceTimersByTime(500) + respond(last("GET"), {status, token: "token"}) + expect(events).toEqual(["open", "suspend", "error", "close"]) + }) + + it("closes when the session is gone", () => { + open() + respond(last("GET"), {status: 410, token: "new-token", messages: [], last: 0}) + + expect(events).toEqual(["open", "error", "close"]) + }) + + it("does not retry after it was closed", () => { + open() + respond(last("GET"), null) + longpoll.close() + const polls = requests("GET").length + jest.advanceTimersByTime(10_000) + + expect(requests("GET")).toHaveLength(polls) + }) + + it("numbers the messages it sends and sends a batch again when it gets no response", () => { + open() + longpoll.send("a") + longpoll.send("b") + jest.advanceTimersByTime(0) + expect(params(last("POST")).seq).toBe("1") + expect(last("POST")[3]).toBe("a\nb") + + longpoll.send("c") + respond(last("POST"), null) + jest.advanceTimersByTime(100) + expect(requests("POST")).toHaveLength(2) + expect(params(last("POST")).seq).toBe("1") + expect(last("POST")[3]).toBe("a\nb") + + respond(last("POST"), {status: 200, seq: 1}) + expect(params(last("POST")).seq).toBe("3") + expect(last("POST")[3]).toBe("c") + }) + + // TODO: Remove in Phoenix 1.10, together with degrade() + it("does not number the messages when the server does not number its messages", () => { + open({status: 410, token: "token", messages: []}) + longpoll.send("a") + jest.advanceTimersByTime(0) + + expect(params(last("POST")).seq).toBeUndefined() + }) + + it("sends messages pushed by a resume callback after a POST recovers", () => { + open() + longpoll.onresume = () => longpoll.send("b") + longpoll.send("a") + jest.advanceTimersByTime(0) + respond(last("POST"), null) + jest.advanceTimersByTime(100) + respond(last("POST"), null) + jest.advanceTimersByTime(500) + respond(last("POST"), {status: 200, seq: 1}) + jest.advanceTimersByTime(0) + expect(last("POST")[3]).toBe("b") + expect(params(last("POST")).seq).toBe("2") + }) + + // During a rolling deploy, a node with an older server forwards our requests + // without ack and seq, so the session treats us like an older client. + // + // TODO: Remove in Phoenix 1.10, together with degrade() + describe("with a node of an older server", () => { + it("delivers the messages of a poll and stops resuming", () => { + open() + respond(last("GET"), {status: 200, token: "token", messages: ["a"]}) + jest.advanceTimersByTime(0) + expect(events).toEqual(["open", "a"]) + expect(params(last("GET")).ack).toBeUndefined() + + // the next failure reconnects, as the session no longer keeps messages for us + respond(last("GET"), null) + expect(events).toEqual(["open", "a", "error", "close"]) + }) + + it.each([ + ["with messages", {status: 200, token: "token", messages: ["a"]}], + ["without messages", {status: 204, token: "token"}] + ])("reconnects when a retried poll %s was answered by it", (_, resp) => { + open() + respond(last("GET"), null) + jest.advanceTimersByTime(100) + // the session dropped the messages of the response we lost + respond(last("GET"), resp) + jest.advanceTimersByTime(0) + expect(events).toEqual(["open", "error", "close"]) + }) + + it("stops resuming after it answered a POST", () => { + open() + longpoll.send("a") + jest.advanceTimersByTime(0) + respond(last("POST"), {status: 200}) + longpoll.send("b") + jest.advanceTimersByTime(0) + expect(params(last("POST")).seq).toBeUndefined() + + respond(last("POST"), null) + expect(events).toEqual(["open", "error", "close"]) + }) + + it("keeps the downgrade when an outstanding modern poll completes", () => { + open() + longpoll.send("a") + jest.advanceTimersByTime(0) + respond(last("POST"), {status: 200}) + respond(last("GET"), {status: 204, token: "token", last: 0}) + expect(params(last("GET")).ack).toBeUndefined() + + longpoll.send("b") + jest.advanceTimersByTime(0) + expect(params(last("POST")).seq).toBeUndefined() + respond(last("GET"), null) + expect(events).toEqual(["open", "error", "close"]) + }) + + it("fails a pending POST retry when a legacy poll downgrades the session", () => { + open() + longpoll.send("a") + jest.advanceTimersByTime(0) + respond(last("POST"), null) + respond(last("GET"), {status: 204, token: "token"}) + jest.advanceTimersByTime(1000) + + expect(requests("POST")).toHaveLength(1) + expect(longpoll.retryTimers.size).toBe(0) + expect(events).toEqual(["open", "error", "close"]) + }) + + it("fails a pending GET retry when a legacy POST downgrades the session", () => { + open() + longpoll.send("a") + jest.advanceTimersByTime(0) + respond(last("GET"), null) + const polls = requests("GET").length + respond(last("POST"), {status: 200}) + jest.advanceTimersByTime(1000) + + expect(requests("GET")).toHaveLength(polls) + expect(longpoll.retryTimers.size).toBe(0) + expect(events).toEqual(["open", "error", "close"]) + }) + + it("keeps resuming while POSTs are answered with seq", () => { + open() + longpoll.send("a") + jest.advanceTimersByTime(0) + respond(last("POST"), {status: 200, seq: 1}) + longpoll.send("b") + jest.advanceTimersByTime(0) + expect(params(last("POST")).seq).toBe("2") + + respond(last("POST"), null) + expect(events).toEqual(["open"]) + }) + }) +}) + describe("Socket with LongPoll", () => { describe("transportConnect", () => { it("should initialize with undefined protocols when no auth token", () => { @@ -849,3 +1215,78 @@ describe("Ajax.request", () => { }) }) }) + +describe("LongPoll continuing a session of another transport", () => { + let longpoll, events + + const requests = method => Ajax.request.mock.calls.filter(([m]) => m === method) + const last = method => requests(method).at(-1) + const params = call => Object.fromEntries(new URL(call[1]).searchParams) + const respond = (call, resp) => call[6](resp) + + const start = (session = {token: "token", received: 3, confirmed: 1, unconfirmed: ["b", "c"]}) => { + longpoll = new LongPoll("ws://localhost/socket/websocket?vsn=2.0.0") + longpoll.resumeSession(session) + events = [] + longpoll.onopen = () => events.push("open") + longpoll.onmessage = ({data}) => events.push(data) + longpoll.onerror = () => events.push("error") + longpoll.onclose = event => events.push({close: event.code}) + jest.advanceTimersByTime(0) + } + + beforeEach(() => { + jest.useFakeTimers() + jest.spyOn(Ajax, "request").mockImplementation(() => ({abort: jest.fn()})) + }) + + afterEach(() => { + longpoll.close() + jest.useRealTimers() + jest.restoreAllMocks() + }) + + it("resumes the session with its first poll", () => { + start() + const poll = last("GET") + expect(poll[1]).toMatch(/^http:\/\/localhost\/socket\/longpoll\?/) + expect(params(poll)).toEqual({vsn: "2.0.0", ack: "3", resume: "1"}) + expect(poll[2]["X-Phoenix-Longpoll-Token"]).toBe("token") + expect(longpoll.readyState).toBe(SOCKET_STATES.connecting) + }) + + it("sends the messages the server did not confirm once it resumed", () => { + start() + longpoll.send("d") + jest.advanceTimersByTime(0) + expect(requests("POST")).toHaveLength(0) + + respond(last("GET"), {status: 200, token: "token", messages: ["x"], last: 4}) + expect(events).toEqual(["open"]) + jest.advanceTimersByTime(0) + expect(events).toEqual(["open", "x"]) + expect(params(last("POST")).seq).toBe("2") + expect(last("POST")[3]).toBe("b\nc\nd") + expect(params(last("GET")).resume).toBeUndefined() + }) + + it("closes when the session is gone", () => { + start() + respond(last("GET"), {status: 410, token: "new-token", messages: [], last: 0}) + expect(events).toEqual(["error", {close: 3410}]) + }) + + // TODO: Remove in Phoenix 1.10, together with degrade() + it("closes when a node with an older server answers", () => { + start() + respond(last("GET"), {status: 204, token: "token"}) + expect(events).toEqual(["error", {close: 3410}]) + }) + + it("delivers binary messages", () => { + start() + respond(last("GET"), {status: 200, token: "token", messages: ["base64:AQID"], last: 4}) + jest.advanceTimersByTime(0) + expect(new Uint8Array(events[1])).toEqual(new Uint8Array([1, 2, 3])) + }) +}) diff --git a/assets/test/resumable_websocket_test.js b/assets/test/resumable_websocket_test.js new file mode 100644 index 0000000000..6b6407cf34 --- /dev/null +++ b/assets/test/resumable_websocket_test.js @@ -0,0 +1,402 @@ +import {jest} from "@jest/globals" +import {ResumableWebSocket, Socket} from "../js/phoenix" +import {SOCKET_STATES} from "../js/phoenix/constants" +import Ajax from "../js/phoenix/ajax" + +const control = (event, payload) => JSON.stringify([null, null, "phoenix", event, payload]) + +class FakeWebSocket { + constructor(url, protocols){ + this.url = url + this.protocols = protocols + this.readyState = SOCKET_STATES.connecting + this.sent = [] + this.closed = null + FakeWebSocket.instances.push(this) + } + + get params(){ return Object.fromEntries(new URL(this.url).searchParams) } + + send(data){ this.sent.push(data) } + + close(code, reason){ + this.closed = {code, reason} + this.readyState = SOCKET_STATES.closed + } + + // the server side + open(){ + this.readyState = SOCKET_STATES.open + this.onopen() + } + + receive(data){ this.onmessage({data}) } + + drop(code = 1006){ + this.readyState = SOCKET_STATES.closed + this.onclose({code, reason: "", wasClean: false}) + } +} + +FakeWebSocket.instances = [] + +const last = () => FakeWebSocket.instances.at(-1) + +describe("ResumableWebSocket", () => { + let transport, events + + const connect = ({session = true} = {}) => { + transport = new ResumableWebSocket("ws://localhost/socket/websocket?vsn=2.0.0", ["phoenix"]) + events = [] + transport.onopen = () => events.push("open") + transport.onmessage = ({data}) => events.push(data) + transport.onerror = () => events.push("error") + transport.onclose = event => events.push({close: event.code}) + transport.onsuspend = () => events.push("suspend") + transport.onresume = () => events.push("resume") + last().open() + if(session){ last().receive(control("phx_session", {token: "token"})) } + } + + // drops the WebSocket and resumes the session over a new one + const resume = (seq, {delay = 100} = {}) => { + last().drop() + jest.advanceTimersByTime(delay) + last().open() + last().receive(control("phx_resumed", {seq})) + } + + beforeEach(() => { + jest.useFakeTimers() + FakeWebSocket.instances = [] + ResumableWebSocket.WebSocket = FakeWebSocket + }) + + afterEach(() => { + transport.close() + ResumableWebSocket.WebSocket = null + jest.useRealTimers() + }) + + it("asks the server for a session", () => { + connect() + expect(last().params).toEqual({vsn: "2.0.0", resumable: "1"}) + expect(last().protocols).toEqual(["phoenix"]) + expect(transport.readyState).toBe(SOCKET_STATES.open) + expect(transport.token).toBe("token") + expect(events).toEqual(["open"]) + }) + + it("delivers messages and acknowledges them", () => { + connect() + last().receive("a") + last().receive("b") + expect(events).toEqual(["open", "a", "b"]) + expect(last().sent).toEqual([]) + + jest.advanceTimersByTime(1000) + expect(last().sent).toEqual([control("phx_ack", {ack: 2})]) + }) + + it("resumes the session over a new WebSocket", () => { + connect() + last().receive("a") + last().drop() + expect(transport.readyState).toBe(SOCKET_STATES.open) + + jest.advanceTimersByTime(99) + expect(FakeWebSocket.instances).toHaveLength(1) + jest.advanceTimersByTime(1) + expect(last().params).toEqual({vsn: "2.0.0", resumable: "1", resume: "1"}) + + last().open() + expect(last().sent).toEqual([control("phx_resume", {token: "token", ack: 1})]) + last().receive(control("phx_resumed", {seq: 0})) + last().receive("b") + expect(events).toEqual(["open", "a", "b"]) + }) + + it("sends the messages the server did not receive again", () => { + connect() + transport.send("a") + transport.send("b") + last().receive(control("phx_ack", {seq: 1})) + expect(transport.unconfirmed).toEqual(["b"]) + last().drop() + + // we keep the messages we send while there is no WebSocket + transport.send("c") + jest.advanceTimersByTime(100) + last().open() + last().receive(control("phx_resumed", {seq: 1})) + expect(last().sent.slice(1)).toEqual(["b", "c"]) + + // the server received "b" before the WebSocket failed + resume(2) + expect(last().sent.slice(1)).toEqual(["c"]) + }) + + it("reports suspension once resuming failed, and resumption once it succeeded", () => { + connect() + last().drop() + jest.advanceTimersByTime(100) + expect(events).toEqual(["open"]) + + last().drop() + expect(events).toEqual(["open", "suspend"]) + jest.advanceTimersByTime(500) + last().open() + last().receive(control("phx_resumed", {seq: 0})) + expect(events).toEqual(["open", "suspend", "resume"]) + }) + + it("backs off while resuming keeps failing", () => { + connect() + const delays = [] + for(let i = 0; i < 6; i++){ + const count = FakeWebSocket.instances.length + last().drop() + let waited = 0 + while(FakeWebSocket.instances.length === count){ + jest.advanceTimersByTime(100) + waited += 100 + } + delays.push(waited) + } + expect(delays).toEqual([100, 500, 1000, 2000, 5000, 5000]) + }) + + it("retries when the server does not answer the resume", () => { + connect() + transport.timeout = 1000 + last().drop() + jest.advanceTimersByTime(100) + const ws = last() + ws.open() + jest.advanceTimersByTime(1000) + expect(ws.closed).not.toBeNull() + jest.advanceTimersByTime(500) + expect(last()).not.toBe(ws) + expect(last().params.resume).toBe("1") + }) + + it("closes with the code of the session when it stopped", () => { + connect() + last().receive(control("phx_gone", {code: 1001})) + expect(events).toEqual(["open", {close: 1001}]) + expect(transport.readyState).toBe(SOCKET_STATES.closed) + }) + + it.each([ + ["the session is gone", ws => ws.receive(control("phx_gone", {}))], + ["an older server replies to the resume", ws => ws.receive(control("phx_reply", {status: "error", response: {}}))], + ["the server treats the resume as a new connection", ws => ws.receive("a")] + ])("closes so the socket connects again when %s", (_, receive) => { + connect() + last().drop() + jest.advanceTimersByTime(100) + const ws = last() + ws.open() + receive(ws) + expect(events).toEqual(["open", {close: 3410}]) + expect(ws.closed).not.toBeNull() + jest.advanceTimersByTime(10_000) + expect(last()).toBe(ws) + }) + + it("closes when the server claims messages we did not send", () => { + connect() + transport.send("a") + last().drop() + jest.advanceTimersByTime(100) + last().open() + last().receive(control("phx_resumed", {seq: 2})) + expect(events).toEqual(["open", {close: 3410}]) + }) + + it("tells the server when it closes", () => { + connect() + const ws = last() + transport.close(1000, "bye") + expect(ws.sent).toEqual([control("phx_close", {})]) + expect(ws.closed).toEqual({code: 1000, reason: "bye"}) + expect(events).toEqual(["open", {close: 1000}]) + }) + + it("does not resume after it was closed", () => { + connect() + last().drop() + transport.close() + jest.advanceTimersByTime(10_000) + expect(FakeWebSocket.instances).toHaveLength(1) + }) + + it("does not retry when a suspension callback closes it", () => { + connect() + transport.onsuspend = () => transport.close() + last().drop() + jest.advanceTimersByTime(100) + last().drop() + jest.advanceTimersByTime(10_000) + expect(FakeWebSocket.instances).toHaveLength(2) + }) + + describe("with a long polling fallback", () => { + const polls = () => Ajax.request.mock.calls.filter(([method]) => method === "GET") + const posts = () => Ajax.request.mock.calls.filter(([method]) => method === "POST") + const params = call => Object.fromEntries(new URL(call[1]).searchParams) + + beforeEach(() => { + jest.spyOn(Ajax, "request").mockImplementation(() => ({abort: jest.fn()})) + }) + + afterEach(() => jest.restoreAllMocks()) + + const interrupt = () => { + connect() + transport.longPollFallbackMs = 2000 + last().receive("a") + transport.send("x") + last().drop() + } + + it("continues the session with long polling when resuming takes too long", () => { + interrupt() + jest.advanceTimersByTime(100) + last().drop() + jest.advanceTimersByTime(1900) + expect(polls()).toHaveLength(0) + const ws = last() + + jest.advanceTimersByTime(100) + expect(ws.closed).not.toBeNull() + jest.advanceTimersByTime(0) + expect(params(polls()[0])).toEqual({vsn: "2.0.0", ack: "1", resume: "1"}) + expect(polls()[0][2]["X-Phoenix-Longpoll-Token"]).toBe("token") + + // the session resumed over long polling + polls()[0][6]({status: 200, token: "token", messages: ["b"], last: 2}) + expect(events).toEqual(["open", "a", "suspend", "resume"]) + jest.advanceTimersByTime(0) + expect(events).toEqual(["open", "a", "suspend", "resume", "b"]) + expect(params(posts()[0]).seq).toBe("1") + expect(posts()[0][3]).toBe("x") + + // we keep long polling and do not open WebSockets anymore + transport.send("y") + jest.advanceTimersByTime(10_000) + expect(FakeWebSocket.instances).toHaveLength(3) + }) + + it.each([ + {status: 200, messages: ["b"], last: 2}, + {status: 204, messages: [], last: 1} + ])("keeps replay order when a resume callback pushes after a $status poll", response => { + interrupt() + transport.onresume = () => transport.send("z") + jest.advanceTimersByTime(100) + last().drop() + jest.advanceTimersByTime(1901) + transport.send("y") + + polls()[0][6]({...response, token: "token"}) + jest.advanceTimersByTime(0) + + expect(posts()).toHaveLength(1) + expect(params(posts()[0]).seq).toBe("1") + expect(posts()[0][3]).toBe("x\ny\nz") + }) + + it("does not fall back once the session resumed over a WebSocket", () => { + interrupt() + jest.advanceTimersByTime(100) + last().open() + last().receive(control("phx_resumed", {seq: 1})) + jest.advanceTimersByTime(10_000) + expect(polls()).toHaveLength(0) + }) + + it("closes when the session is gone", () => { + interrupt() + // the first poll runs right after we fall back + jest.advanceTimersByTime(2001) + polls()[0][6]({status: 410, token: "new", messages: [], last: 0}) + expect(events).toEqual(["open", "a", "error", {close: 3410}]) + expect(transport.readyState).toBe(SOCKET_STATES.closed) + }) + + it("closes the LongPoll when it closes", () => { + interrupt() + // the first poll runs right after we fall back + jest.advanceTimersByTime(2001) + const longpoll = transport.longpoll + transport.close(1000, "bye") + expect(longpoll.readyState).toBe(SOCKET_STATES.closed) + expect(events.at(-1)).toEqual({close: 1000}) + }) + }) + + describe("with a server that does not keep sessions", () => { + it("behaves like a WebSocket", () => { + connect({session: false}) + transport.send("a") + last().receive("b") + expect(transport.unconfirmed).toEqual([]) + expect(last().sent).toEqual(["a"]) + + jest.advanceTimersByTime(1000) + expect(last().sent).toEqual(["a"]) + + last().drop(1006) + expect(events).toEqual(["open", "b", {close: 1006}]) + jest.advanceTimersByTime(10_000) + expect(FakeWebSocket.instances).toHaveLength(1) + }) + + it("reports errors of the WebSocket", () => { + connect({session: false}) + last().onerror("boom") + expect(events).toEqual(["open", "error"]) + }) + }) +}) + +describe("Socket with resumable: true", () => { + let socket + + beforeEach(() => { + jest.useFakeTimers() + FakeWebSocket.instances = [] + ResumableWebSocket.WebSocket = FakeWebSocket + }) + + afterEach(() => { + socket.disconnect() + ResumableWebSocket.WebSocket = null + jest.useRealTimers() + }) + + it("resumes the session when the heartbeat times out", () => { + socket = new Socket("ws://localhost/socket", {resumable: true, heartbeatIntervalMs: 1000}) + expect(socket.transportName(socket.transport)).toBe("ResumableWebSocket") + socket.connect() + last().open() + last().receive(control("phx_session", {token: "token"})) + const channel = socket.channel("room") + const errors = [] + channel.onError(() => errors.push("error")) + channel.join() + + // the heartbeat is sent, but its reply never arrives + jest.advanceTimersByTime(1000) + const ws = last() + jest.advanceTimersByTime(1000) + expect(ws.closed).not.toBeNull() + expect(errors).toEqual([]) + + jest.advanceTimersByTime(100) + expect(last()).not.toBe(ws) + expect(last().params.resume).toBe("1") + expect(socket.conn).toBeInstanceOf(ResumableWebSocket) + }) +}) diff --git a/assets/test/socket_test.js b/assets/test/socket_test.js index af2e2f5759..59c99f56a6 100644 --- a/assets/test/socket_test.js +++ b/assets/test/socket_test.js @@ -60,7 +60,7 @@ describe("with transports", function (){ expect(socket.sendBuffer.length).toBe(0) expect(socket.ref).toBe(0) expect(socket.endPoint).toBe("/socket/websocket") - expect(socket.stateChangeCallbacks).toEqual({open: [], close: [], error: [], message: []}) + expect(socket.stateChangeCallbacks).toEqual({open: [], close: [], error: [], message: [], suspend: [], resume: []}) expect(socket.transport).toBe(WebSocket) expect(socket.timeout).toBe(10000) expect(socket.longpollerTimeout).toBe(20000) diff --git a/assets/test/suspend_test.js b/assets/test/suspend_test.js new file mode 100644 index 0000000000..cc42337aca --- /dev/null +++ b/assets/test/suspend_test.js @@ -0,0 +1,84 @@ +import {jest} from "@jest/globals" +import {Socket} from "../js/phoenix" +import {SOCKET_STATES} from "../js/phoenix/constants" + +class Transport { + constructor(){ this.readyState = SOCKET_STATES.connecting } + send(){} + close(){ this.readyState = SOCKET_STATES.closed } + open(){ this.readyState = SOCKET_STATES.open; this.onopen() } +} + +describe("transport suspension", () => { + let socket, channel, transport, events + + beforeEach(() => { + jest.useFakeTimers() + events = [] + socket = new Socket("/socket", {transport: Transport}) + socket.onSuspend(() => events.push("suspend")) + socket.onResume(() => events.push("resume")) + socket.connect() + transport = socket.conn + transport.open() + channel = socket.channel("room") + channel.onError(() => events.push("error")) + channel.join() + transport.onmessage({data: JSON.stringify([ + channel.joinRef(), channel.joinRef(), "room", "phx_reply", {status: "ok", response: {}} + ])}) + }) + + afterEach(() => { + socket.disconnect() + jest.useRealTimers() + }) + + it("preserves the join and delivers one notification per transition", () => { + const joinRef = channel.joinRef() + transport.onsuspend() + transport.onsuspend() + expect(events).toEqual(["suspend"]) + expect(channel.canPush()).toBe(true) + transport.onresume() + transport.onresume() + expect(events).toEqual(["suspend", "resume"]) + expect(channel.joinRef()).toBe(joinRef) + }) + + it("errors the channel if the suspended transport loses its session", () => { + transport.onsuspend() + transport.close() + transport.onclose({code: 3410}) + transport.onresume() + expect(events).toEqual(["suspend", "error"]) + expect(channel.isErrored()).toBe(true) + }) + + it("ignores events from a replaced transport", () => { + transport.onsuspend() + socket.disconnect() + socket.connect() + socket.conn.open() + events = [] + transport.onsuspend() + transport.onresume() + expect(events).toEqual([]) + }) + + it("allows removing callbacks with off", () => { + const callback = jest.fn() + socket.off([socket.onSuspend(callback), socket.onResume(callback)]) + transport.onsuspend() + transport.onresume() + expect(callback).not.toHaveBeenCalled() + }) + + it("stops notifying when a callback disconnects", () => { + socket.onSuspend(() => socket.disconnect()) + const callback = jest.fn() + socket.onSuspend(callback) + transport.onsuspend() + expect(callback).not.toHaveBeenCalled() + }) +}) diff --git a/lib/phoenix.ex b/lib/phoenix.ex index 58cc2771be..f1e86e2a44 100644 --- a/lib/phoenix.ex +++ b/lib/phoenix.ex @@ -29,7 +29,7 @@ defmodule Phoenix do children = [ # Code reloading must be serial across all Phoenix apps Phoenix.CodeReloader.Server, - {DynamicSupervisor, name: Phoenix.Transports.LongPoll.Supervisor, strategy: :one_for_one} + {DynamicSupervisor, name: Phoenix.Transports.Session.Supervisor, strategy: :one_for_one} ] Supervisor.start_link(children, strategy: :one_for_one, name: Phoenix.Supervisor) diff --git a/lib/phoenix/endpoint.ex b/lib/phoenix/endpoint.ex index c2ad6a182b..8c47104808 100644 --- a/lib/phoenix/endpoint.ex +++ b/lib/phoenix/endpoint.ex @@ -735,7 +735,10 @@ defmodule Phoenix.Endpoint do :error_handler, :validate_utf8, :active_n, - :deflate_options + :deflate_options, + :resumable, + :window_ms, + :crypto ] ) @@ -1084,6 +1087,47 @@ defmodule Phoenix.Endpoint do def handle_error(conn, :rate_limit), do: Plug.Conn.send_resp(conn, 429, "Too many requests") + * `:resumable` - whether clients can resume the socket over a new WebSocket + after the connection failed, without rejoining channels. Defaults to `false`. + Clients opt in with the `resumable: true` option of the `phoenix.js` Socket, + which requires the V2 serializer. Each socket then runs in its own process, + which keeps the messages the client did not acknowledge yet and stays alive for + `:window_ms` times 1.5 after the WebSocket failed. A resumed socket does not invoke + `c:Phoenix.Socket.connect/3`. Implement `c:Phoenix.Socket.resume/3` to check + the new connection instead. When the socket also enables `:longpoll`, clients + with the `longPollFallbackMs` option continue the socket with long polling + when they cannot resume it over a WebSocket in time. This requires the same + `:crypto` options for both transports. + + To decide for each connection, pass a function as MFA tuple instead of `true`. + It runs after `c:Phoenix.Socket.connect/3` accepted a new connection and must + return a boolean. It receives the map given to + `c:Phoenix.Socket.Transport.connect/1`, with the `:params` and `:connect_info` + of the connection, followed by the arguments of the tuple. When it returns + `false`, the client connects like it does to a socket that is not resumable. + For example, `mix phx.gen.auth` stores a `"user_token"` in the session of + signed-in users, so this only lets them resume: + + socket "/socket", MySocket, + websocket: [ + connect_info: [session: @session_options], + resumable: {MySocket, :resumable?, []} + ] + + with `MySocket` defining: + + def resumable?(%{connect_info: %{session: %{"user_token" => _}}}), do: true + def resumable?(_transport_info), do: false + + The example does not verify the token, as it only decides whether the socket + is resumable. The function does not run when a client resumes. + + * `:window_ms` - with `:resumable`, how long the client can take to resume + the socket in milliseconds (ms), times 1.5. Defaults to `10_000`. + + * `:crypto` - with `:resumable`, options for verifying and signing the resume + token, accepted by `Phoenix.Token`. By default tokens are valid for 2 weeks + ## Longpoll configuration The following configuration applies only to `:longpoll`: diff --git a/lib/phoenix/socket.ex b/lib/phoenix/socket.ex index c47f532c02..1bb60650ef 100644 --- a/lib/phoenix/socket.ex +++ b/lib/phoenix/socket.ex @@ -249,7 +249,30 @@ defmodule Phoenix.Socket do """ @callback id(Socket.t()) :: String.t() | nil - @optional_callbacks connect: 2, connect: 3 + @doc """ + Checks a client that resumes its socket over a new connection. + + Long polling, and WebSockets with the `:resumable` option, keep the socket and + its channels when a request or connection fails, and the client resumes them + over a new one. Only `c:connect/3` checks the first + connection, so this callback lets you check the new one, for example to verify + that the user is still logged in. It receives the params and `connect_info` of + the new connection, and the socket returned by `c:connect/3`, with the + `:transport` of the new connection. + + Return `{:ok, socket}` to resume. The returned assigns apply to channels joined + afterwards. Channels that already joined keep their sockets. The socket id cannot change. + + To deny the resume, return `:error` or `{:error, term}`. The socket and its + channels stop, and the client connects again and rejoins its channels, + which invokes `c:connect/3`. + + If the callback is not implemented, the client resumes the socket. + """ + @callback resume(params :: map, Socket.t(), connect_info :: map) :: + {:ok, Socket.t()} | {:error, term} | :error + + @optional_callbacks connect: 2, connect: 3, resume: 3 defmodule InvalidMessageError do @moduledoc """ @@ -327,6 +350,9 @@ defmodule Phoenix.Socket do @doc false def handle_info(message, state), do: Phoenix.Socket.__info__(message, state) + @doc false + def handle_resume(info, state), do: Phoenix.Socket.__resume__(info, state) + @doc false def terminate(reason, state), do: Phoenix.Socket.__terminate__(reason, state) end @@ -604,6 +630,34 @@ defmodule Phoenix.Socket do {:ok, state} end + def __resume__(info, {state, socket}) do + %{transport: transport, params: params, connect_info: connect_info} = info + %{handler: handler, id: id} = socket = %{socket | transport: transport} + + result = + if function_exported?(handler, :resume, 3) do + handler.resume(params, socket, connect_info) + else + {:ok, socket} + end + + case result do + {:ok, %Socket{} = socket} -> + {:ok, {state, %{socket | id: id}}} + + :error -> + {:error, :rejected} + + {:error, reason} -> + {:error, reason} + + invalid -> + raise ArgumentError, + "#{inspect(handler)}.resume/3 must return {:ok, socket}, :error " <> + "or {:error, term}, got: #{inspect(invalid)}" + end + end + def __terminate__(_reason, _state_socket) do :ok end diff --git a/lib/phoenix/socket/transport.ex b/lib/phoenix/socket/transport.ex index 966fc7d942..5c9bcaad5c 100644 --- a/lib/phoenix/socket/transport.ex +++ b/lib/phoenix/socket/transport.ex @@ -235,6 +235,21 @@ defmodule Phoenix.Socket.Transport do | {:push, {opcode :: atom, message :: term}, state} | {:stop, reason :: term, state} + @doc """ + Handles a client that resumes its session over a new connection. + + Transports that keep the socket across connections, such as long polling and + resumable WebSockets, invoke it when a client resumes after a failed request or + connection, before the client receives the messages it missed. It receives a map with the same + keys as `c:connect/1`, describing the new connection. It must return one of: + + * `{:ok, state}` - continues the socket + * `{:error, reason}` - stops the socket, so the client connects again + + If the callback is not implemented, the client resumes the socket. + """ + @callback handle_resume(transport_info :: map, state) :: {:ok, state} | {:error, term()} + @doc """ Invoked on termination. @@ -244,7 +259,7 @@ defmodule Phoenix.Socket.Transport do """ @callback terminate(reason :: term, state) :: :ok - @optional_callbacks handle_control: 2, drainer_spec: 1 + @optional_callbacks handle_control: 2, handle_resume: 2, drainer_spec: 1 require Logger diff --git a/lib/phoenix/transports/long_poll.ex b/lib/phoenix/transports/long_poll.ex index 4085e1abf7..84f0433058 100644 --- a/lib/phoenix/transports/long_poll.ex +++ b/lib/phoenix/transports/long_poll.ex @@ -11,6 +11,7 @@ defmodule Phoenix.Transports.LongPoll do import Plug.Conn alias Phoenix.Socket.{V1, V2, Transport} + alias Phoenix.Transports.Session def default_config() do [ @@ -54,8 +55,8 @@ defmodule Phoenix.Transports.LongPoll do # Starts a new session or listen to a message if one already exists. defp dispatch(%{method: "GET"} = conn, endpoint, handler, opts) do case resume_session(conn, endpoint, opts) do - {:ok, new_conn, server_ref, token} -> - listen(new_conn, server_ref, token, endpoint, opts) + {:ok, new_conn, server_ref, priv_topic, token} -> + listen(new_conn, {server_ref, priv_topic}, token, endpoint, opts) :error -> new_session(conn, endpoint, handler, opts) @@ -65,7 +66,7 @@ defmodule Phoenix.Transports.LongPoll do # Publish the message. defp dispatch(%{method: "POST"} = conn, endpoint, _, opts) do case resume_session(conn, endpoint, opts) do - {:ok, new_conn, server_ref, _token} -> + {:ok, new_conn, server_ref, _priv_topic, _token} -> publish(new_conn, server_ref, endpoint, opts) :error -> @@ -82,27 +83,39 @@ defmodule Phoenix.Transports.LongPoll do case read_body(conn, []) do {:ok, body, conn} -> # We need to match on both v1 and v2 protocol, as well as wrap for backwards compat - status = + {status, seq} = case get_req_header(conn, "content-type") do ["application/x-ndjson"] -> - body - |> String.splitter(["\n", "\r\n"]) - |> Stream.take(@max_poll_batch_size) - |> Enum.find_value(fn part -> - msg = - case part do - "[" <> _ = txt -> {txt, :text} - base64 -> {safe_decode64!(base64), :binary} - end - - transport_dispatch(endpoint, server_ref, msg, opts) - end) + # the seq of the first message, sent by clients that resend messages + # + # TODO: Require seq in Phoenix 1.10. Phoenix 1.8 clients don't send it. + # Phoenix 1.9 clients only omit it after their session reached a 1.8 node. + seq = int_param(conn, "seq") + + status = + body + |> String.splitter(["\n", "\r\n"]) + |> Stream.take(@max_poll_batch_size) + |> Stream.with_index() + |> Enum.find_value(fn {part, index} -> + msg = + case part do + "[" <> _ = txt -> {txt, :text} + base64 -> {safe_decode64!(base64), :binary} + end + + transport_dispatch(endpoint, server_ref, msg, seq && seq + index, opts) + end) + + {status, seq} _ -> - transport_dispatch(endpoint, server_ref, {body, :text}, opts) + {transport_dispatch(endpoint, server_ref, {body, :text}, nil, opts), nil} end - conn |> put_status(status || :ok) |> status_json() + # the seq tells clients that we drop the messages they send again + extra = if seq, do: %{"seq" => seq}, else: %{} + conn |> put_status(status || :ok) |> status_json(extra) _ -> raise Plug.BadRequestError @@ -117,9 +130,16 @@ defmodule Phoenix.Transports.LongPoll do end end - defp transport_dispatch(endpoint, server_ref, body, opts) do + defp transport_dispatch(endpoint, server_ref, body, seq, opts) do ref = make_ref() - broadcast_from!(endpoint, server_ref, {:dispatch, client_ref(server_ref), body, ref}) + + # TODO: Always send seq in Phoenix 1.10, see publish/4 + message = + if seq, + do: {:dispatch, client_ref(server_ref), body, ref, seq}, + else: {:dispatch, client_ref(server_ref), body, ref} + + broadcast_from!(endpoint, server_ref, message) receive do {:ok, ^ref} -> nil @@ -132,10 +152,7 @@ defmodule Phoenix.Transports.LongPoll do ## Session handling defp new_session(conn, endpoint, handler, opts) do - priv_topic = - "phx:lp:" <> - Base.encode64(:crypto.strong_rand_bytes(16)) <> - (System.system_time(:millisecond) |> Integer.to_string()) + priv_topic = Session.priv_topic() keys = Keyword.get(opts, :connect_info, []) @@ -145,48 +162,126 @@ defmodule Phoenix.Transports.LongPoll do Transport.connect_info(conn, endpoint, keys, Keyword.take(opts, @connect_info_opts)) arg = {endpoint, handler, opts, conn.params, priv_topic, connect_info} - spec = {Phoenix.Transports.LongPoll.Server, arg} + spec = {Session, arg} - case DynamicSupervisor.start_child(Phoenix.Transports.LongPoll.Supervisor, spec) do + case DynamicSupervisor.start_child(Session.Supervisor, spec) do :ignore -> conn |> put_status(:forbidden) |> status_json() {:ok, server_pid} -> data = {:v1, endpoint.config(:endpoint_id), server_pid, priv_topic} - token = sign_token(endpoint, data, opts) - conn |> put_status(:gone) |> status_token_messages_json(token, []) + token = Session.sign_token(endpoint, data, opts) + # the last message tells clients that we keep messages until they acknowledge them + conn |> put_status(:gone) |> status_token_messages_json(token, [], 0) + end + end + + defp listen(conn, {server_ref, priv_topic}, token, endpoint, opts) do + case check_resume(conn, server_ref, priv_topic, endpoint, opts) do + :ok -> + poll(conn, server_ref, token, endpoint, opts) + + :rejected -> + # the session is gone, so the client starts a new one + conn |> put_status(:gone) |> status_json() end end - defp listen(conn, server_ref, token, endpoint, opts) do + # Clients resume the session with the first poll after a failed request, + # so the socket can check the connection again. + defp check_resume(%{params: %{"resume" => _}} = conn, server_ref, priv_topic, endpoint, opts) do + keys = Keyword.get(opts, :connect_info, []) + conn = maybe_auth_token_from_header(conn, opts[:auth_token]) + + connect_info = + Transport.connect_info(conn, endpoint, keys, Keyword.take(opts, @connect_info_opts)) + + info = %{ + endpoint: endpoint, + transport: :longpoll, + options: opts, + params: Map.drop(conn.params, ["ack", "resume"]), + connect_info: connect_info + } + + ref = make_ref() + resume = {:resume, client_ref(server_ref), ref, info, priv_topic} + broadcast_from!(endpoint, server_ref, resume) + + receive do + {:resumed, ^ref} -> :ok + {:rejected, ^ref} -> :rejected + after + # The session handles the resume before the flush we send next, as both come from + # this process. If it is slow to answer, for example because resume/3 queries the + # database, we poll anyway: a session that rejects the resume stops before it sends + # us messages, so the poll expires and the next request gets 410. + opts[:pubsub_timeout_ms] -> :ok + end + end + + defp check_resume(_conn, _server_ref, _priv_topic, _endpoint, _opts), do: :ok + + defp poll(conn, server_ref, token, endpoint, opts) do ref = make_ref() client_ref = client_ref(server_ref) - broadcast_from!(endpoint, server_ref, {:flush, client_ref, ref}) - {status, messages} = + # the last message the client received, sent by clients that acknowledge + ack = int_param(conn, "ack") + + flush = + case ack do + # TODO: Remove the nil case in Phoenix 1.10. Phoenix 1.8 clients don't send + # an ack. Phoenix 1.9 clients only omit it after their session reached a 1.8 node. + nil -> {:flush, client_ref, ref} + ack -> {:flush, client_ref, ref, ack} + end + + broadcast_from!(endpoint, server_ref, flush) + + {status, messages, last} = receive do + # Phoenix 1.8 servers, and newer servers answering a flush without ack, + # reply without the number of the last message. + # + # TODO: Remove the clauses without last in Phoenix 1.10 {:messages, messages, ^ref} -> - {:ok, messages} + {:ok, messages, nil} + + {:messages, messages, ^ref, last} -> + {:ok, messages, last} {:now_available, ^ref} -> - broadcast_from!(endpoint, server_ref, {:flush, client_ref, ref}) + broadcast_from!(endpoint, server_ref, flush) receive do - {:messages, messages, ^ref} -> {:ok, messages} + {:messages, messages, ^ref} -> {:ok, messages, nil} + {:messages, messages, ^ref, last} -> {:ok, messages, last} after opts[:window_ms] -> broadcast_from!(endpoint, server_ref, {:expired, client_ref, ref}) - {:no_content, []} + {:no_content, [], ack} end after opts[:window_ms] -> + # Without messages, the last message is the one the client acknowledged. + # It tells the client that we keep the messages, like we do with messages. broadcast_from!(endpoint, server_ref, {:expired, client_ref, ref}) - {:no_content, []} + {:no_content, [], ack} end conn |> put_status(status) - |> status_token_messages_json(token, messages) + |> status_token_messages_json(token, messages, last) + end + + defp int_param(conn, key) do + with value when is_binary(value) <- conn.params[key], + {int, ""} <- Integer.parse(value) do + int + else + _ -> nil + end end # Accept the token from either location so that newer clients can use the @@ -198,11 +293,11 @@ defmodule Phoenix.Transports.LongPoll do end end - # Retrieves the serialized `Phoenix.LongPoll.Server` pid + # Retrieves the serialized `Phoenix.Transports.Session` pid # by publishing a message in the encrypted private topic. defp resume_session(%Plug.Conn{} = conn, endpoint, opts) do with token when is_binary(token) <- fetch_token(conn), - {:ok, {:v1, id, pid, priv_topic}} <- verify_token(endpoint, token, opts) do + {:ok, {:v1, id, pid, priv_topic}} <- Session.verify_token(endpoint, token, opts) do server_ref = server_ref(endpoint.config(:endpoint_id), id, pid, priv_topic) new_conn = @@ -216,7 +311,7 @@ defmodule Phoenix.Transports.LongPoll do broadcast_from!(endpoint, server_ref, {:subscribe, client_ref(server_ref), ref}) receive do - {:subscribe, ^ref} -> {:ok, new_conn, server_ref, token} + {:subscribe, ^ref} -> {:ok, new_conn, server_ref, priv_topic, token} after opts[:pubsub_timeout_ms] -> :error end @@ -256,24 +351,6 @@ defmodule Phoenix.Transports.LongPoll do defp broadcast_from!(_endpoint, pid, msg) when is_pid(pid), do: send(pid, msg) - defp sign_token(endpoint, data, opts) do - Phoenix.Token.sign( - endpoint, - Atom.to_string(endpoint.config(:pubsub_server)), - data, - opts[:crypto] - ) - end - - defp verify_token(endpoint, signed, opts) do - Phoenix.Token.verify( - endpoint, - Atom.to_string(endpoint.config(:pubsub_server)), - signed, - opts[:crypto] - ) - end - defp maybe_auth_token_from_header(conn, true) do case Plug.Conn.get_req_header(conn, "x-phoenix-authtoken") do [] -> @@ -286,14 +363,24 @@ defmodule Phoenix.Transports.LongPoll do defp maybe_auth_token_from_header(conn, _), do: conn - defp status_json(conn) do - send_json(conn, %{"status" => conn.status || 200}) + defp status_json(conn, extra \\ %{}) do + send_json(conn, Map.put(extra, "status", conn.status || 200)) end - defp status_token_messages_json(conn, token, messages) do + # TODO: Remove in Phoenix 1.10, when every poll has a last message + defp status_token_messages_json(conn, token, messages, nil) do send_json(conn, %{"status" => conn.status || 200, "token" => token, "messages" => messages}) end + defp status_token_messages_json(conn, token, messages, last) do + send_json(conn, %{ + "status" => conn.status || 200, + "token" => token, + "messages" => messages, + "last" => last + }) + end + defp send_json(conn, data) do conn |> put_resp_header("content-type", "application/json; charset=utf-8") diff --git a/lib/phoenix/transports/long_poll_server.ex b/lib/phoenix/transports/long_poll_server.ex deleted file mode 100644 index e3a5f23d8d..0000000000 --- a/lib/phoenix/transports/long_poll_server.ex +++ /dev/null @@ -1,161 +0,0 @@ -defmodule Phoenix.Transports.LongPoll.Server do - @moduledoc false - - use GenServer, restart: :temporary - alias Phoenix.PubSub - - def start_link(arg) do - GenServer.start_link(__MODULE__, arg) - end - - def init({endpoint, handler, options, params, priv_topic, connect_info}) do - config = %{ - endpoint: endpoint, - transport: :longpoll, - options: options, - params: params, - connect_info: connect_info - } - - window_ms = Keyword.fetch!(options, :window_ms) - - case handler.connect(config) do - {:ok, handler_state} -> - {:ok, handler_state} = handler.init(handler_state) - - state = %{ - buffer: [], - handler: {handler, handler_state}, - window_ms: trunc(window_ms * 1.5), - pubsub_server: endpoint.config(:pubsub_server), - priv_topic: priv_topic, - last_client_poll: now_ms(), - client_ref: nil - } - - :ok = PubSub.subscribe(state.pubsub_server, priv_topic) - schedule_inactive_shutdown(state.window_ms) - {:ok, state} - - :error -> - :ignore - - {:error, _reason} -> - :ignore - end - end - - def handle_info({:dispatch, client_ref, {body, opcode}, ref}, state) do - %{handler: {handler, handler_state}} = state - - case handler.handle_in({body, opcode: opcode}, handler_state) do - {:reply, status, {_, reply}, handler_state} -> - state = %{state | handler: {handler, handler_state}} - status = if status == :ok, do: :ok, else: :error - broadcast_from!(state, client_ref, {status, ref}) - publish_reply(state, reply) - - {:ok, handler_state} -> - state = %{state | handler: {handler, handler_state}} - broadcast_from!(state, client_ref, {:ok, ref}) - {:noreply, state} - - {:stop, reason, handler_state} -> - state = %{state | handler: {handler, handler_state}} - broadcast_from!(state, client_ref, {:error, ref}) - {:stop, reason, state} - end - end - - def handle_info({:subscribe, client_ref, ref}, state) do - broadcast_from!(state, client_ref, {:subscribe, ref}) - {:noreply, state} - end - - def handle_info({:flush, client_ref, ref}, state) do - case state.buffer do - [] -> - {:noreply, %{state | client_ref: {client_ref, ref}, last_client_poll: now_ms()}} - - buffer -> - broadcast_from!(state, client_ref, {:messages, Enum.reverse(buffer), ref}) - {:noreply, %{state | client_ref: nil, last_client_poll: now_ms(), buffer: []}} - end - end - - def handle_info({:expired, client_ref, ref}, state) do - case state.client_ref do - {^client_ref, ^ref} -> - {:noreply, %{state | client_ref: nil}} - - _ -> - {:noreply, state} - end - end - - def handle_info(:shutdown_if_inactive, state) do - if now_ms() - state.last_client_poll > state.window_ms do - {:stop, {:shutdown, :inactive}, state} - else - schedule_inactive_shutdown(state.window_ms) - {:noreply, state} - end - end - - def handle_info(message, state) do - %{handler: {handler, handler_state}} = state - - case handler.handle_info(message, handler_state) do - {:push, {_, reply}, handler_state} -> - state = %{state | handler: {handler, handler_state}} - publish_reply(state, reply) - - {:ok, handler_state} -> - state = %{state | handler: {handler, handler_state}} - {:noreply, state} - - {:stop, reason, handler_state} -> - state = %{state | handler: {handler, handler_state}} - {:stop, reason, state} - end - end - - def terminate(reason, state) do - %{handler: {handler, handler_state}} = state - handler.terminate(reason, handler_state) - :ok - end - - defp broadcast_from!(state, client_ref, msg) when is_binary(client_ref), - do: PubSub.broadcast_from!(state.pubsub_server, self(), client_ref, msg) - - defp broadcast_from!(_state, client_ref, msg) when is_pid(client_ref), - do: send(client_ref, msg) - - defp publish_reply(state, reply) when is_map(reply) do - IO.warn( - "Returning a map from the LongPolling serializer is deprecated. " <> - "Please return JSON encoded data instead (see Phoenix.Socket.Serializer)" - ) - - publish_reply(state, Phoenix.json_library().encode_to_iodata!(reply)) - end - - defp publish_reply(state, reply) do - notify_client_now_available(state) - {:noreply, update_in(state.buffer, &[IO.iodata_to_binary(reply) | &1])} - end - - defp notify_client_now_available(state) do - case state.client_ref do - {client_ref, ref} -> broadcast_from!(state, client_ref, {:now_available, ref}) - nil -> :ok - end - end - - defp now_ms, do: System.system_time(:millisecond) - - defp schedule_inactive_shutdown(window_ms) do - Process.send_after(self(), :shutdown_if_inactive, window_ms) - end -end diff --git a/lib/phoenix/transports/session.ex b/lib/phoenix/transports/session.ex new file mode 100644 index 0000000000..a88a1e1a94 --- /dev/null +++ b/lib/phoenix/transports/session.ex @@ -0,0 +1,500 @@ +defmodule Phoenix.Transports.Session do + @moduledoc false + # + # A socket session that outlives the requests or connections of its transport. + # + # The session runs the socket handler, so it is the `transport_pid` of the socket + # and its channels. Transports talk to it through short-lived processes, such as + # the request processes of long polling or the process of a WebSocket, so a client + # can resume the session after one of them failed. A WebSocket attaches to the + # session as its pipe, which receives our messages as we send them, while long + # polling requests them. + # + # Both directions number their messages, starting at 1: + # + # * The session numbers the messages it sends and keeps them until the client + # acknowledges them with the number of the last message it received. It sends + # all messages after the acknowledged one again, together with the number of + # the last one, so the client can drop the ones it already received. + # + # * The client numbers the messages it sends. The session drops the ones it + # already received, as the client sends them again when it did not get our + # response. Messages must arrive in order, as a gap means we lost messages. + # + # The numbers do not depend on how a transport groups messages, so a client can + # resume the session with another request or connection than the one that failed. + # + # Long polling addresses the session with a `client_ref`, which is the pid of the + # request process when it runs on a node we are connected to, or the PubSub + # topic it subscribed to otherwise. Pipes talk to the session directly, so they + # must run on a node we are connected to. + + use GenServer, restart: :temporary + alias Phoenix.PubSub + + # the WebSocket close codes we use when the session stops, see RFC 6455 + @close_codes %{normal: 1000, internal_error: 1011, service_restart: 1012} + + @doc """ + Starts a session for a socket that its transport connected already. + """ + def start(endpoint, handler, handler_state, options) do + priv_topic = priv_topic() + arg = {:connected, endpoint, handler, handler_state, options, priv_topic} + + case DynamicSupervisor.start_child(__MODULE__.Supervisor, {__MODULE__, arg}) do + {:ok, pid} -> {:ok, pid, priv_topic} + {:error, _} = error -> error + end + end + + @doc false + def priv_topic do + "phx:lp:" <> + Base.encode64(:crypto.strong_rand_bytes(16)) <> + (System.system_time(:millisecond) |> Integer.to_string()) + end + + @doc """ + Signs the token that clients use to resume the session. + """ + def sign_token(endpoint, data, opts) do + Phoenix.Token.sign( + endpoint, + Atom.to_string(endpoint.config(:pubsub_server)), + data, + opts[:crypto] + ) + end + + @doc """ + Verifies a token signed with `sign_token/3`. + """ + def verify_token(endpoint, signed, opts) do + Phoenix.Token.verify( + endpoint, + Atom.to_string(endpoint.config(:pubsub_server)), + signed, + opts[:crypto] + ) + end + + def start_link(arg) do + GenServer.start_link(__MODULE__, arg) + end + + def init({:connected, endpoint, handler, handler_state, options, priv_topic}) do + {:ok, handler_state} = handler.init(handler_state) + {:ok, init_state(endpoint, handler, handler_state, options, priv_topic)} + end + + def init({endpoint, handler, options, params, priv_topic, connect_info}) do + config = %{ + endpoint: endpoint, + transport: :longpoll, + options: options, + params: params, + connect_info: connect_info + } + + case handler.connect(config) do + {:ok, handler_state} -> + {:ok, handler_state} = handler.init(handler_state) + {:ok, init_state(endpoint, handler, handler_state, options, priv_topic)} + + :error -> + :ignore + + {:error, _reason} -> + :ignore + end + end + + defp init_state(endpoint, handler, handler_state, options, priv_topic) do + window_ms = Keyword.fetch!(options, :window_ms) + + state = %{ + handler: {handler, handler_state}, + # the messages the client did not acknowledge yet, as {seq, opcode, payload} + out: :queue.new(), + # the number of the last message we numbered, sent and the client acknowledged + out_seq: 0, + sent: 0, + acked: 0, + # when we sent the oldest message the client did not acknowledge yet + sent_at: nil, + # the number of the last message we received + in_seq: 0, + window_ms: trunc(window_ms * 1.5), + pubsub_server: endpoint.config(:pubsub_server), + priv_topic: priv_topic, + last_client_poll: now_ms(), + client_ref: nil, + # the process of a connection that receives our messages as we send them, + # such as a WebSocket, and its monitor + pipe: nil, + # the close code the socket asked for when it stopped + close_code: nil + } + + :ok = PubSub.subscribe(state.pubsub_server, priv_topic) + schedule_inactive_shutdown(state.window_ms) + state + end + + ## Incoming messages + + def handle_info({:dispatch, client_ref, _body, ref, seq}, %{in_seq: in_seq} = state) + when seq <= in_seq do + # The client sends messages again when it did not get the response to its request. + broadcast_from!(state, client_ref, {:ok, ref}) + {:noreply, state} + end + + def handle_info({:dispatch, client_ref, _body, ref, seq}, %{in_seq: in_seq} = state) + when seq > in_seq + 1 do + # We cannot process the message, as we lost the ones before it. + broadcast_from!(state, client_ref, {:error, ref}) + {:stop, {:shutdown, :out_of_order}, state} + end + + def handle_info({:dispatch, client_ref, body, ref, _seq}, state) do + dispatch(client_ref, body, ref, state) + end + + # TODO: Remove in Phoenix 1.10. Phoenix 1.8 nodes and clients dispatch without seq. + def handle_info({:dispatch, client_ref, body, ref}, state) do + dispatch(client_ref, body, ref, state) + end + + def handle_info({:subscribe, client_ref, ref}, state) do + broadcast_from!(state, client_ref, {:subscribe, ref}) + {:noreply, state} + end + + # Transports find the session by the pid in the token. A pid can belong to another + # session after this one stopped, so a client that resumes also sends the topic of + # the token, which only this session has. We reject a mismatch but keep running for + # our own client. + def handle_info({:resume, client_ref, ref, _info, priv_topic}, state) + when priv_topic != state.priv_topic do + broadcast_from!(state, client_ref, {:rejected, ref}) + {:noreply, state} + end + + # A client that resumes the session after a failed request or connection + # tells us about its new connection, so the socket can check it. + def handle_info({:resume, client_ref, ref, info, _priv_topic}, state) do + %{handler: {handler, handler_state}} = state + + result = + if function_exported?(handler, :handle_resume, 2) do + handler.handle_resume(info, handler_state) + else + {:ok, handler_state} + end + + case result do + {:ok, handler_state} -> + broadcast_from!(state, client_ref, {:resumed, ref}) + # the client gave up on the WebSocket we have, for example to long poll instead + {:noreply, %{detach(state) | handler: {handler, handler_state}}} + + {:error, _reason} -> + broadcast_from!(state, client_ref, {:rejected, ref}) + {:stop, {:shutdown, :resume_rejected}, state} + end + end + + ## Pipes + + # A connection that receives our messages as we send them, such as a WebSocket. + # It replaces the previous one, which the client gave up on. + def handle_info({:pipe_attach, pipe, ref, _ack, priv_topic}, state) + when priv_topic != state.priv_topic do + send(pipe, {:rejected, ref}) + {:noreply, state} + end + + def handle_info({:pipe_attach, pipe, ref, ack, _priv_topic}, state) do + # the client resumed over a WebSocket, so it stopped polling + state = %{detach(state) | client_ref: nil} + send(pipe, {:attached, ref, state.in_seq}) + state = acknowledge(state, ack) + state = %{state | pipe: {pipe, Process.monitor(pipe)}} + + if :queue.is_empty(state.out) do + {:noreply, state} + else + send( + pipe, + {:push_frames, + for({_seq, opcode, payload} <- :queue.to_list(state.out), do: {opcode, payload})} + ) + + {:noreply, sent(state)} + end + end + + # Pipes number the messages they receive, starting after the last one we received + # when they attached. The pipe the client gave up on can still deliver messages, + # which the client sends again over the new one, so we drop them like duplicates. + def handle_info({:pipe_in, pipe, seq, _frame}, %{in_seq: in_seq} = state) when seq <= in_seq do + send(pipe, {:in_seq, in_seq}) + {:noreply, state} + end + + def handle_info({:pipe_in, _pipe, seq, _frame}, %{in_seq: in_seq} = state) + when seq > in_seq + 1 do + {:stop, {:shutdown, :out_of_order}, %{state | close_code: @close_codes.internal_error}} + end + + def handle_info({:pipe_in, pipe, _seq, {body, opcode}}, state) do + %{handler: {handler, handler_state}} = state + state = %{state | in_seq: state.in_seq + 1} + send(pipe, {:in_seq, state.in_seq}) + + case handler.handle_in({body, opcode: opcode}, handler_state) do + {:reply, _status, {opcode, reply}, handler_state} -> + publish_reply(%{state | handler: {handler, handler_state}}, opcode, reply) + + {:ok, handler_state} -> + {:noreply, %{state | handler: {handler, handler_state}}} + + {:stop, reason, handler_state} -> + {:stop, reason, %{state | handler: {handler, handler_state}}} + end + end + + def handle_info({:pipe_ack, _pipe, ack}, state) do + {:noreply, acknowledge(state, ack)} + end + + # The client closes the session, as opposed to losing its connection. + def handle_info({:pipe_close, pipe}, %{pipe: {pipe, _}} = state) do + {:stop, {:shutdown, :closed}, %{state | close_code: @close_codes.normal}} + end + + def handle_info({:pipe_close, _pipe}, state) do + {:noreply, state} + end + + def handle_info({:DOWN, ref, _, _, _}, %{pipe: {_, ref}} = state) do + {:noreply, %{state | pipe: nil, last_client_poll: now_ms()}} + end + + ## Outgoing messages + + # The client acknowledges the last message it received with each poll, so we send + # the messages after it until it did, as the response of a poll can get lost. + def handle_info({:flush, client_ref, ref, ack}, state) do + state = acknowledge(%{state | last_client_poll: now_ms()}, ack) + + cond do + # A poll the client gave up on when it resumed over a WebSocket. It must not + # take the messages, so it expires. A client that switches to long polling + # resumes the session first, which detaches the WebSocket. + state.pipe != nil -> + {:noreply, state} + + :queue.is_empty(state.out) -> + {:noreply, %{state | client_ref: {client_ref, ref}}} + + true -> + broadcast_from!(state, client_ref, {:messages, poll_payloads(state), ref, state.out_seq}) + {:noreply, %{sent(state) | client_ref: nil}} + end + end + + # Backwards compat: clients that don't acknowledge get each message once. + # + # TODO: Remove in Phoenix 1.10. Phoenix 1.8 nodes and clients flush without ack. + def handle_info({:flush, client_ref, ref}, state) do + # we assume the client received the messages we sent before, like it does + state = acknowledge(%{state | last_client_poll: now_ms()}, state.sent) + + if :queue.is_empty(state.out) do + {:noreply, %{state | client_ref: {client_ref, ref}}} + else + broadcast_from!(state, client_ref, {:messages, poll_payloads(state), ref}) + {:noreply, %{acknowledge(sent(state), state.out_seq) | client_ref: nil}} + end + end + + def handle_info({:expired, client_ref, ref}, state) do + case state.client_ref do + {^client_ref, ^ref} -> + {:noreply, %{state | client_ref: nil}} + + _ -> + {:noreply, state} + end + end + + def handle_info(:shutdown_if_inactive, state) do + cond do + state.pipe == nil and now_ms() - state.last_client_poll > state.window_ms -> + {:stop, {:shutdown, :inactive}, state} + + unacknowledged?(state) -> + {:stop, {:shutdown, :unacknowledged}, state} + + true -> + schedule_inactive_shutdown(state.window_ms) + {:noreply, state} + end + end + + def handle_info(message, state) do + %{handler: {handler, handler_state}} = state + + case handler.handle_info(message, handler_state) do + {:push, {opcode, reply}, handler_state} -> + state = %{state | handler: {handler, handler_state}} + publish_reply(state, opcode, reply) + + {:ok, handler_state} -> + state = %{state | handler: {handler, handler_state}} + {:noreply, state} + + {:stop, reason, handler_state} -> + state = %{state | handler: {handler, handler_state}} + {:stop, reason, state} + + {:stop, reason, code, handler_state} -> + state = %{state | handler: {handler, handler_state}, close_code: code} + {:stop, reason, state} + end + end + + def terminate(reason, state) do + %{handler: {handler, handler_state}} = state + + with {pipe, _ref} <- state.pipe do + send(pipe, {:session_closed, state.close_code || close_code(reason)}) + end + + handler.terminate(reason, handler_state) + :ok + end + + defp close_code(:normal), do: @close_codes.normal + defp close_code({:shutdown, :restart}), do: @close_codes.service_restart + defp close_code(_reason), do: @close_codes.internal_error + + defp detach(%{pipe: {pipe, ref}} = state) do + Process.demonitor(ref, [:flush]) + send(pipe, {:detached, self()}) + %{state | pipe: nil, last_client_poll: now_ms()} + end + + defp detach(state), do: state + + defp dispatch(client_ref, {body, opcode}, ref, state) do + %{handler: {handler, handler_state}} = state + state = %{state | in_seq: state.in_seq + 1} + + case handler.handle_in({body, opcode: opcode}, handler_state) do + {:reply, status, {opcode, reply}, handler_state} -> + state = %{state | handler: {handler, handler_state}} + status = if status == :ok, do: :ok, else: :error + broadcast_from!(state, client_ref, {status, ref}) + publish_reply(state, opcode, reply) + + {:ok, handler_state} -> + state = %{state | handler: {handler, handler_state}} + broadcast_from!(state, client_ref, {:ok, ref}) + {:noreply, state} + + {:stop, reason, handler_state} -> + state = %{state | handler: {handler, handler_state}} + broadcast_from!(state, client_ref, {:error, ref}) + {:stop, reason, state} + end + end + + # Long polling sends text messages as they are and binary messages in base64 + # with a prefix, which a session can have when it started over a WebSocket. + defp poll_payloads(state) do + for {_seq, opcode, payload} <- :queue.to_list(state.out) do + if opcode == :binary, do: "base64:" <> Base.encode64(payload), else: payload + end + end + + # Drops the messages up to the acknowledged one. A late request can acknowledge + # an older message, and a client cannot acknowledge messages we did not send. + defp acknowledge(state, ack) when is_integer(ack) and ack > state.acked do + ack = min(ack, state.sent) + out = drop_until(state.out, ack) + # while the client makes progress, the rest of the messages get a new window + sent_at = if ack < state.sent, do: now_ms() + %{state | out: out, acked: ack, sent_at: sent_at} + end + + defp acknowledge(state, _ack), do: state + + defp drop_until(queue, ack) do + case :queue.peek(queue) do + {:value, {seq, _opcode, _payload}} when seq <= ack -> drop_until(:queue.drop(queue), ack) + _ -> queue + end + end + + defp sent(state) do + %{state | sent: state.out_seq, sent_at: state.sent_at || now_ms()} + end + + defp broadcast_from!(state, client_ref, msg) when is_binary(client_ref), + do: PubSub.broadcast_from!(state.pubsub_server, self(), client_ref, msg) + + defp broadcast_from!(_state, client_ref, msg) when is_pid(client_ref), + do: send(client_ref, msg) + + defp publish_reply(state, opcode, reply) when is_map(reply) do + IO.warn( + "Returning a map from the LongPolling serializer is deprecated. " <> + "Please return JSON encoded data instead (see Phoenix.Socket.Serializer)" + ) + + publish_reply(state, opcode, Phoenix.json_library().encode_to_iodata!(reply)) + end + + defp publish_reply(state, opcode, reply) do + # A client that does not acknowledge must not be able to + # grow the buffer indefinitely, so we close when it did not acknowledge + # the messages we sent within the window. + if unacknowledged?(state) do + {:stop, {:shutdown, :unacknowledged}, state} + else + seq = state.out_seq + 1 + payload = IO.iodata_to_binary(reply) + state = %{state | out: :queue.in({seq, opcode, payload}, state.out), out_seq: seq} + + case state.pipe do + {pipe, _ref} -> + send(pipe, {:push_frames, [{opcode, payload}]}) + {:noreply, sent(state)} + + nil -> + notify_client_now_available(state) + {:noreply, state} + end + end + end + + defp unacknowledged?(%{sent_at: nil}), do: false + defp unacknowledged?(state), do: now_ms() - state.sent_at > state.window_ms + + defp notify_client_now_available(state) do + case state.client_ref do + {client_ref, ref} -> broadcast_from!(state, client_ref, {:now_available, ref}) + nil -> :ok + end + end + + defp now_ms, do: System.monotonic_time(:millisecond) + + defp schedule_inactive_shutdown(window_ms) do + Process.send_after(self(), :shutdown_if_inactive, window_ms) + end +end diff --git a/lib/phoenix/transports/websocket.ex b/lib/phoenix/transports/websocket.ex index 6859663a6e..49e192a13a 100644 --- a/lib/phoenix/transports/websocket.ex +++ b/lib/phoenix/transports/websocket.ex @@ -22,6 +22,10 @@ defmodule Phoenix.Transports.WebSocket do import Plug.Conn alias Phoenix.Socket.{V1, V2, Transport} + alias Phoenix.Transports.{Session, WebSocket.Pipe} + + # the params of resumable clients, which we don't pass to the socket + @resume_params ["resumable", "resume"] def default_config() do [ @@ -30,7 +34,10 @@ defmodule Phoenix.Transports.WebSocket do error_handler: {__MODULE__, :handle_error, []}, timeout: 60_000, transport_log: false, - compress: false + compress: false, + resumable: false, + window_ms: 10_000, + crypto: [max_age: 1_209_600] ] end @@ -68,32 +75,99 @@ defmodule Phoenix.Transports.WebSocket do endpoint: endpoint, transport: :websocket, options: opts, - params: params, + params: Map.drop(params, @resume_params), connect_info: connect_info } - case handler.connect(config) do - {:ok, arg} -> - try do - conn - |> WebSockAdapter.upgrade(handler, arg, opts) - |> halt() - rescue - e in WebSockAdapter.UpgradeError -> send_resp(conn, 400, e.message) - end - - :error -> - send_resp(conn, 403, "") - - {:error, reason} -> - {m, f, args} = opts[:error_handler] - apply(m, f, [conn, reason | args]) + case resumable(params, opts) do + :resume -> + upgrade(conn, Pipe, {:resume, endpoint, opts, config}, opts) + + :new -> + connect(conn, handler, config, opts, fn arg -> + if resumable?(opts[:resumable], config) do + start_session(conn, endpoint, handler, arg, opts) + else + upgrade(conn, handler, arg, opts) + end + end) + + nil -> + connect(conn, handler, config, opts, &upgrade(conn, handler, &1, opts)) end end end def call(conn, _), do: send_resp(conn, 400, "") + # Resumable clients ask for a session that outlives the WebSocket. Only clients + # with the V2 serializer can resume, as the session talks to them with V2 messages. + defp resumable(%{"resumable" => _} = params, opts) do + # a function given as :resumable decides for each new connection, see resumable?/2 + with resumable when resumable != false <- Keyword.get(opts, :resumable, false), + "2." <> _ <- params["vsn"] do + if params["resume"], do: :resume, else: :new + else + _ -> nil + end + end + + defp resumable(_params, _opts), do: nil + + # Whether a new connection gets a session. Clients that resume don't get here, as + # they have the token of a session we started, which the socket checks with resume/3. + defp resumable?(true, _config), do: true + + defp resumable?({module, function, args}, config) do + case apply(module, function, [config | args]) do + resumable when is_boolean(resumable) -> + resumable + + invalid -> + raise ArgumentError, + "#{inspect(module)}.#{function}/#{length(args) + 1}, given as :resumable, " <> + "must return a boolean, got: #{inspect(invalid)}" + end + end + + defp resumable?(invalid, _config) do + raise ArgumentError, ":resumable expects a boolean or MFA tuple, got: #{inspect(invalid)}" + end + + defp start_session(conn, endpoint, handler, arg, opts) do + case Session.start(endpoint, handler, arg, opts) do + {:ok, session, priv_topic} -> + data = {:v1, endpoint.config(:endpoint_id), session, priv_topic} + token = Session.sign_token(endpoint, data, opts) + upgrade(conn, Pipe, {:new, session, priv_topic, token, opts}, opts) + + {:error, _} -> + send_resp(conn, 500, "") + end + end + + defp connect(conn, handler, config, opts, upgrade) do + case handler.connect(config) do + {:ok, arg} -> + upgrade.(arg) + + :error -> + send_resp(conn, 403, "") + + {:error, reason} -> + {m, f, args} = opts[:error_handler] + apply(m, f, [conn, reason | args]) + end + end + + defp upgrade(conn, handler, arg, opts) do + conn + |> WebSockAdapter.upgrade(handler, arg, opts) + |> halt() + rescue + e in WebSockAdapter.UpgradeError -> send_resp(conn, 400, e.message) + end + def handle_error(conn, _reason), do: send_resp(conn, 403, "") defp maybe_auth_token_from_header(%{halted: true} = conn, _), do: conn diff --git a/lib/phoenix/transports/websocket_pipe.ex b/lib/phoenix/transports/websocket_pipe.ex new file mode 100644 index 0000000000..c0a06313e2 --- /dev/null +++ b/lib/phoenix/transports/websocket_pipe.ex @@ -0,0 +1,232 @@ +defmodule Phoenix.Transports.WebSocket.Pipe do + @moduledoc false + # + # The WebSocket of a resumable socket. + # + # The socket runs in a `Phoenix.Transports.Session`, which outlives the WebSocket, so + # the client can resume it over a new WebSocket. The pipe forwards the messages of + # the client to the session, and the messages of the session to the client. + # + # Besides the messages of the socket, the pipe and the client exchange control + # messages, which are V2 messages without refs on the "phoenix" topic: + # + # * "phx_session" - the first message of a new session, with its resume token + # * "phx_resume" - the first message of the client on a new WebSocket, with the + # token and the number of the last message it received + # * "phx_resumed" - the answer to "phx_resume", with the number of the last + # message the session received + # * "phx_ack" - the number of the last message the client or the session received + # * "phx_close" - the client closes the socket, as opposed to losing its WebSocket + # * "phx_gone" - the session stopped or the client cannot resume it, with the + # close code of the session if it stopped + # + # Control messages are not numbered. + + @behaviour WebSock + + alias Phoenix.Transports.Session + + @control_prefix "[null,null,\"phoenix\",\"phx_" + + # the WebSocket close codes we use, see RFC 6455 + @close_codes %{ + normal: 1000, + invalid_payload: 1007, + policy_violation: 1008, + internal_error: 1011 + } + + @impl true + def init({:new, session, priv_topic, token, opts}) do + state = new_state(opts) + + case attach(state, session, priv_topic, 0) do + {:ok, state} -> {:push, [control("phx_session", %{token: token})], state} + :error -> gone(state, @close_codes.internal_error) + end + end + + def init({:resume, endpoint, opts, info}) do + # the client must send the resume message right after connecting + timer = Process.send_after(self(), :resume_timeout, Keyword.fetch!(opts, :window_ms)) + {:ok, %{new_state(opts) | resume: {endpoint, info, timer}}} + end + + defp new_state(opts) do + %{ + opts: opts, + session: nil, + session_ref: nil, + # set until the client resumed the session + resume: nil, + # the number of the last message we forwarded to the session + in_seq: 0, + # the number of the last message the session confirmed, and whether + # we are about to tell the client about it + confirmed: 0, + ack_scheduled: false + } + end + + @impl true + def handle_in({@control_prefix <> _ = text, opcode: :text}, state) do + case Phoenix.json_library().decode(text) do + {:ok, [nil, nil, "phoenix", event, payload]} -> handle_control(event, payload, state) + _ -> {:stop, :normal, @close_codes.invalid_payload, state} + end + end + + def handle_in(_frame, %{resume: {_, _, _}} = state) do + # the client must resume the session before sending messages + gone(state, nil) + end + + def handle_in({payload, opcode: opcode}, state) do + seq = state.in_seq + 1 + send(state.session, {:pipe_in, self(), seq, {payload, opcode}}) + {:ok, %{state | in_seq: seq}} + end + + defp handle_control("phx_resume", %{"token" => token, "ack" => ack}, %{resume: resume} = state) + when is_binary(token) and is_integer(ack) and resume != nil do + {endpoint, info, timer} = resume + Process.cancel_timer(timer) + state = %{state | resume: nil} + + with {:ok, session, priv_topic} <- verify_token(endpoint, token, state.opts), + :ok <- check_resume(session, priv_topic, info, state.opts), + {:ok, state} <- attach(state, session, priv_topic, ack) do + {:push, [control("phx_resumed", %{seq: state.in_seq})], state} + else + _ -> gone(state, nil) + end + end + + defp handle_control("phx_ack", %{"ack" => ack}, %{session: session} = state) + when is_integer(ack) and session != nil do + send(session, {:pipe_ack, self(), ack}) + {:ok, state} + end + + defp handle_control("phx_close", _payload, %{session: session} = state) when session != nil do + send(session, {:pipe_close, self()}) + {:stop, :normal, state} + end + + defp handle_control(_event, _payload, state) do + {:stop, :normal, @close_codes.policy_violation, state} + end + + @impl true + def handle_info({:push_frames, frames}, state) do + {:push, frames, state} + end + + def handle_info({:in_seq, seq}, state) do + # we tell the client once for the messages the session confirmed in a row + state = %{state | confirmed: max(state.confirmed, seq)} + + if state.ack_scheduled do + {:ok, state} + else + send(self(), :send_ack) + {:ok, %{state | ack_scheduled: true}} + end + end + + def handle_info(:send_ack, state) do + {:push, [control("phx_ack", %{seq: state.confirmed})], %{state | ack_scheduled: false}} + end + + def handle_info({:session_closed, code}, state) do + gone(state, code) + end + + def handle_info({:DOWN, ref, _, _, _}, %{session_ref: ref} = state) do + gone(state, @close_codes.internal_error) + end + + def handle_info({:detached, session}, %{session: session} = state) do + # the client resumed the session over another WebSocket + {:stop, :normal, state} + end + + def handle_info(:resume_timeout, %{resume: {_, _, _}} = state) do + {:stop, :normal, @close_codes.policy_violation, state} + end + + def handle_info(_message, state) do + {:ok, state} + end + + defp attach(state, session, priv_topic, ack) do + ref = Process.monitor(session) + send(session, {:pipe_attach, self(), ref, ack, priv_topic}) + + receive do + {:attached, ^ref, in_seq} -> + {:ok, %{state | session: session, session_ref: ref, in_seq: in_seq, confirmed: in_seq}} + + {:rejected, ^ref} -> + Process.demonitor(ref, [:flush]) + :error + + {:DOWN, ^ref, _, _, _} -> + :error + after + Keyword.fetch!(state.opts, :window_ms) -> + Process.demonitor(ref, [:flush]) + :error + end + end + + defp verify_token(endpoint, token, opts) do + case Session.verify_token(endpoint, token, opts) do + {:ok, {:v1, endpoint_id, session, priv_topic}} when is_pid(session) -> + # We talk to the session directly, so it must run on a node we are connected to. + # Pids of this node look the same after a restart, so the endpoint that signed the + # token must still be running, and the session checks the topic as well. + cond do + node(session) in Node.list() -> {:ok, session, priv_topic} + endpoint_id == endpoint.config(:endpoint_id) -> {:ok, session, priv_topic} + true -> :error + end + + _ -> + :error + end + end + + defp check_resume(session, priv_topic, info, opts) do + ref = Process.monitor(session) + send(session, {:resume, self(), ref, info, priv_topic}) + + receive do + {:resumed, ^ref} -> + Process.demonitor(ref, [:flush]) + :ok + + {:rejected, ^ref} -> + Process.demonitor(ref, [:flush]) + :error + + {:DOWN, ^ref, _, _, _} -> + :error + after + Keyword.fetch!(opts, :window_ms) -> + Process.demonitor(ref, [:flush]) + :error + end + end + + # The client connects again with a new session when it cannot resume, + # and closes with the code of the session if it stopped. + defp gone(state, code) do + payload = if code, do: %{code: code}, else: %{} + {:stop, :normal, code || @close_codes.normal, [control("phx_gone", payload)], state} + end + + defp control(event, payload) do + {:text, Phoenix.json_library().encode_to_iodata!([nil, nil, "phoenix", event, payload])} + end +end diff --git a/test/phoenix/integration/long_poll_channels_test.exs b/test/phoenix/integration/long_poll_channels_test.exs index 2513f61b66..17e4434429 100644 --- a/test/phoenix/integration/long_poll_channels_test.exs +++ b/test/phoenix/integration/long_poll_channels_test.exs @@ -110,6 +110,22 @@ defmodule Phoenix.Integration.LongPollChannelsTest do {:ok, assign(socket, :user_id, params["user_id"])} end + def resume(%{"reject_resume" => "true"}, _socket, _connect_info), do: :error + + # answers after the plug stopped waiting, which is 200ms in these tests + def resume(%{"slow_resume" => result} = params, socket, connect_info) do + Process.sleep(300) + params = Map.delete(params, "slow_resume") + + if result == "reject", + do: :error, + else: resume(params, socket, connect_info) + end + + def resume(params, socket, _connect_info) do + {:ok, assign(socket, :resumed_with, params["resumed_with"])} + end + def id(socket) do if id = socket.assigns.user_id, do: "user_sockets:#{id}" end @@ -181,8 +197,8 @@ defmodule Phoenix.Integration.LongPollChannelsTest do end setup config do - for {_, pid, _, _} <- DynamicSupervisor.which_children(Phoenix.Transports.LongPoll.Supervisor) do - DynamicSupervisor.terminate_child(Phoenix.Transports.LongPoll.Supervisor, pid) + for {_, pid, _, _} <- DynamicSupervisor.which_children(Phoenix.Transports.Session.Supervisor) do + DynamicSupervisor.terminate_child(Phoenix.Transports.Session.Supervisor, pid) end {:ok, topic: "room:" <> to_string(config.test)} @@ -193,6 +209,8 @@ defmodule Phoenix.Integration.LongPollChannelsTest do assert_receive {:DOWN, ^ref, :process, _pid, _} end + # TODO: Number the messages in Phoenix 1.10, when the server requires ack and seq. + # The helpers below poll like Phoenix 1.8 clients, which most tests rely on. @doc """ Helper method to maintain token session state when making HTTP requests. @@ -656,10 +674,206 @@ defmodule Phoenix.Integration.LongPollChannelsTest do assert_receive %Broadcast{event: "new_msg", payload: %{"body" => "hi!"}} end + # TODO: Remove in Phoenix 1.10, together with the flush without ack + # in Phoenix.Transports.Session + test "#{@mode}: accepts legacy flushes without waiting for acknowledgements" do + vsn = "2.0.0" + session = join("/ws", "room:lobby", vsn, "1", @mode) + resp = poll(:get, "/ws", vsn, Map.put(session, "ack", "0")) + assert resp.body["status"] == 200 + + {:ok, {:v1, _id, server, _topic}} = + Phoenix.Token.verify(Endpoint, Atom.to_string(__MODULE__), session["token"]) + + # an older plug flushes without acknowledging and expects the reply without last + Endpoint.broadcast!("room:lobby", "new_msg", %{"body" => "hi"}) + Process.sleep(50) + ref = make_ref() + send(server, {:flush, self(), ref}) + assert_receive {:messages, [_new_msg], ^ref} + + # it does not acknowledge the messages we sent before, which we drop + # like we drop the ones we send to it, so we don't wait for an ack + assert %{sent_at: nil, out_seq: 4, acked: 4} = state = :sys.get_state(server) + assert :queue.is_empty(state.out) + end + + test "#{@mode}: resends messages until the client acknowledges them" do + vsn = "2.0.0" + + # new sessions tell clients that they keep messages + resp = poll(:get, "/ws", vsn, %{}) + assert resp.body["status"] == 410 + assert resp.body["last"] == 0 + + session = join("/ws", "room:lobby", vsn, "1", @mode) + resp = poll(:get, "/ws", vsn, Map.put(session, "ack", "0")) + assert resp.body["status"] == 200 + assert [_reply, _user_entered, _joined] = messages = resp.body["messages"] + assert resp.body["last"] == 3 + + # the response got lost, so the client polls with the previous ack + resp = poll(:get, "/ws", vsn, Map.put(session, "ack", "0")) + assert resp.body["status"] == 200 + assert resp.body["messages"] == messages + assert resp.body["last"] == 3 + + # new messages are sent with the ones the client did not acknowledge + Endpoint.broadcast("room:lobby", "new_msg", %{"body" => "hi"}) + Process.sleep(50) + resp = poll(:get, "/ws", vsn, Map.put(session, "ack", "2")) + assert [%Message{event: "joined"}, %Message{event: "new_msg"}] = resp.body["messages"] + assert resp.body["last"] == 4 + + # a late poll acknowledging fewer messages does not bring back dropped ones + resp = poll(:get, "/ws", vsn, Map.put(session, "ack", "1")) + assert [%Message{event: "joined"}, %Message{event: "new_msg"}] = resp.body["messages"] + + # without messages, the last message is the acknowledged one + resp = poll(:get, "/ws", vsn, Map.put(session, "ack", "4")) + assert resp.body["status"] == 204 + assert resp.body["last"] == 4 + end + + test "#{@mode}: ignores acknowledgements of messages we did not send" do + vsn = "2.0.0" + session = join("/ws", "room:lobby", vsn, "1", @mode) + + {:ok, {:v1, _id, server, _topic}} = + Phoenix.Token.verify(Endpoint, Atom.to_string(__MODULE__), session["token"]) + + Process.sleep(50) + ref = make_ref() + send(server, {:flush, self(), ref, 100}) + assert_receive {:messages, [_reply, _user_entered, _joined], ^ref, 3} + end + + test "#{@mode}: closes when the client keeps polling without acknowledging" do + vsn = "2.0.0" + session = join("/ws", "room:lobby", vsn, "1", @mode) + + # the window is 200ms, so the client must acknowledge messages within 300ms + statuses = + for _ <- 1..8 do + Process.sleep(100) + poll(:get, "/ws", vsn, Map.put(session, "ack", "0")).body["status"] + end + + assert [200, 200 | _] = statuses + assert List.last(statuses) == 410 + end + + test "#{@mode}: drops messages the client sends again" do + vsn = "2.0.0" + Phoenix.PubSub.subscribe(__MODULE__, "room:lobby") + session = join("/ws", "room:lobby", vsn, "1", @mode) + + message = fn body -> + %{ + "topic" => "room:lobby", + "event" => "new_msg", + "ref" => body, + "payload" => %{"body" => body} + } + end + + resp = poll(:post, "/ws", vsn, Map.put(session, "seq", "2"), [message.("a")]) + # the seq tells the client that we drop messages it sends again + assert resp.body["status"] == 200 + assert resp.body["seq"] == 2 + assert_receive %Broadcast{event: "new_msg", payload: %{"body" => "a"}} + + # the response got lost, so the client sends the message again with the next one + batch = [message.("a"), message.("b")] + resp = poll(:post, "/ws", vsn, Map.put(session, "seq", "2"), batch) + assert resp.body["status"] == 200 + assert_receive %Broadcast{event: "new_msg", payload: %{"body" => "b"}} + refute_received %Broadcast{event: "new_msg", payload: %{"body" => "a"}} + end + + test "#{@mode}: closes when messages arrive out of order" do + vsn = "2.0.0" + session = join("/ws", "room:lobby", vsn, "1", @mode) + + message = %{ + "topic" => "room:lobby", + "event" => "new_msg", + "ref" => "1", + "payload" => %{"body" => "a"} + } + + # the join was message 1, so message 3 means we lost message 2 + resp = poll(:post, "/ws", vsn, Map.put(session, "seq", "3"), [message]) + assert resp.body["status"] == 200 + + resp = poll(:get, "/ws", vsn, Map.put(session, "ack", "0")) + assert resp.body["status"] == 410 + end + + test "#{@mode}: checks the socket when the client resumes" do + vsn = "2.0.0" + session = join("/ws", "room:lobby", vsn, "1", @mode) + + {:ok, {:v1, _id, server, _topic}} = + Phoenix.Token.verify(Endpoint, Atom.to_string(__MODULE__), session["token"]) + + params = Map.merge(session, %{"ack" => "0", "resume" => "1", "resumed_with" => "a"}) + resp = poll(:get, "/ws", vsn, params) + assert resp.body["status"] == 200 + assert [_reply, _user_entered, _joined] = resp.body["messages"] + + assert {_, %Phoenix.Socket{assigns: %{resumed_with: "a"}, transport: :longpoll}} = + elem(:sys.get_state(server).handler, 1) + + # the channel keeps running + assert Process.whereis(:"room:lobby") + end + + test "#{@mode}: does not send messages when a slow resume is rejected" do + vsn = "2.0.0" + session = join("/ws", "room:lobby", vsn, "1", @mode) + + params = Map.merge(session, %{"ack" => "0", "resume" => "1", "slow_resume" => "reject"}) + resp = poll(:get, "/ws", vsn, params) + assert resp.body["status"] == 204 + assert resp.body["messages"] == [] + assert_down("room:lobby") + + resp = poll(:get, "/ws", vsn, Map.put(session, "ack", "0")) + assert resp.body["status"] == 410 + end + + test "#{@mode}: sends messages when a slow resume is accepted" do + vsn = "2.0.0" + session = join("/ws", "room:lobby", vsn, "1", @mode) + + params = Map.merge(session, %{"ack" => "0", "resume" => "1", "slow_resume" => "accept"}) + resp = poll(:get, "/ws", vsn, params) + assert resp.body["status"] == 200 + assert [_reply, _user_entered, _joined] = resp.body["messages"] + end + + test "#{@mode}: stops the socket when it rejects the resume" do + vsn = "2.0.0" + session = join("/ws", "room:lobby", vsn, "1", @mode) + + params = Map.merge(session, %{"ack" => "0", "resume" => "1", "reject_resume" => "true"}) + resp = poll(:get, "/ws", vsn, params) + assert resp.body["status"] == 410 + refute resp.body["token"] + assert_down("room:lobby") + + # the client starts a new session + resp = poll(:get, "/ws", vsn, Map.put(session, "ack", "0")) + assert resp.body["status"] == 410 + assert resp.body["token"] + end + test "#{@mode}: lonpoll publishing batch events on v2 protocol" do vsn = "2.0.0" Phoenix.PubSub.subscribe(__MODULE__, "room:lobby") session = join("/ws", "room:lobby", vsn, "1", @mode) + # Publish successfully resp = poll(:post, "/ws", vsn, session, [ @@ -686,7 +900,11 @@ defmodule Phoenix.Integration.LongPollChannelsTest do } ]) + # clients that don't number their messages get no seq + # + # TODO: Remove in Phoenix 1.10, when the server requires seq assert resp.body["status"] == 200 + refute Map.has_key?(resp.body, "seq") assert_receive %Broadcast{event: "new_msg", payload: %{"body" => "hi1"}} assert_receive %Broadcast{event: "new_msg", payload: %{"body" => "hi2"}} refute_receive %Broadcast{event: "new_msg", payload: %{"body" => "stale"}} diff --git a/test/phoenix/integration/long_poll_socket_test.exs b/test/phoenix/integration/long_poll_socket_test.exs index 88b7eec69d..1fe6a8ef89 100644 --- a/test/phoenix/integration/long_poll_socket_test.exs +++ b/test/phoenix/integration/long_poll_socket_test.exs @@ -84,8 +84,8 @@ defmodule Phoenix.Integration.LongPollSocketTest do end setup do - for {_, pid, _, _} <- DynamicSupervisor.which_children(Phoenix.Transports.LongPoll.Supervisor) do - DynamicSupervisor.terminate_child(Phoenix.Transports.LongPoll.Supervisor, pid) + for {_, pid, _, _} <- DynamicSupervisor.which_children(Phoenix.Transports.Session.Supervisor) do + DynamicSupervisor.terminate_child(Phoenix.Transports.Session.Supervisor, pid) end :ok diff --git a/test/phoenix/integration/websocket_resume_test.exs b/test/phoenix/integration/websocket_resume_test.exs new file mode 100644 index 0000000000..496bb5a2d7 --- /dev/null +++ b/test/phoenix/integration/websocket_resume_test.exs @@ -0,0 +1,532 @@ +Code.require_file("../../support/websocket_client.exs", __DIR__) +Code.require_file("../../support/http_client.exs", __DIR__) + +defmodule Phoenix.Integration.WebSocketResumeTest do + # TODO: use parameterized tests once we require Elixir 1.18 + use ExUnit.Case + + import ExUnit.CaptureLog + + alias Phoenix.Integration.{HTTPClient, WebsocketClient} + alias Phoenix.Socket.{Message, V2} + alias __MODULE__.Endpoint + + @moduletag :capture_log + @port 5909 + + Application.put_env(:phoenix, Endpoint, + https: false, + http: [port: @port], + secret_key_base: String.duplicate("abcdefgh", 8), + server: true, + drainer: false, + pubsub_server: __MODULE__ + ) + + defmodule RoomChannel do + use Phoenix.Channel + + def join(topic, _message, socket) do + Process.register(self(), String.to_atom(topic)) + send(self(), :after_join) + {:ok, socket} + end + + def handle_info(:after_join, socket) do + push(socket, "joined", %{transport: inspect(socket.transport)}) + {:noreply, socket} + end + + def handle_info({:push, event, payload}, socket) do + push(socket, event, payload) + {:noreply, socket} + end + + def handle_in("echo", payload, socket) do + push(socket, "echo", payload) + {:noreply, socket} + end + + def handle_in("binary", _payload, socket) do + push(socket, "binary", {:binary, <<1, 2, 3>>}) + {:noreply, socket} + end + end + + defmodule UserSocket do + use Phoenix.Socket + + channel "room:*", RoomChannel + + def connect(%{"reject_connect" => "true"}, _socket, _connect_info), do: :error + + def connect(params, socket, _connect_info) do + send(Process.whereis(:resume_test), {:connect, params}) + {:ok, assign(socket, :user_id, params["user_id"])} + end + + def resume(%{"reject_resume" => "true"}, _socket, _connect_info), do: :error + + def resume(params, socket, _connect_info) do + send(Process.whereis(:resume_test), {:resume, params}) + {:ok, socket} + end + + def id(socket) do + if id = socket.assigns.user_id, do: "user_sockets:#{id}" + end + + # the :resumable function of "/ws/users", which only lets users resume + def resumable?(%{params: params} = transport_info, arg) do + send(Process.whereis(:resume_test), {:resumable?, transport_info, arg}) + + case params["user_id"] do + "invalid" -> :invalid + user_id -> is_binary(user_id) + end + end + end + + defmodule Endpoint do + use Phoenix.Endpoint, otp_app: :phoenix + + socket "/ws", UserSocket, + websocket: [resumable: true, window_ms: 200], + longpoll: [window_ms: 200, pubsub_timeout_ms: 200] + + socket "/ws/plain", UserSocket, websocket: true + + socket "/ws/users", UserSocket, + websocket: [resumable: {UserSocket, :resumable?, [:arg]}, window_ms: 200] + end + + setup %{adapter: adapter} do + Process.register(self(), :resume_test) + config = Application.get_env(:phoenix, Endpoint) + Application.put_env(:phoenix, Endpoint, Keyword.merge(config, adapter: adapter)) + capture_log(fn -> start_supervised!(Endpoint) end) + start_supervised!({Phoenix.PubSub, name: __MODULE__}) + :ok + end + + defp connect(params, path \\ "/ws") do + query = URI.encode_query(Map.merge(%{"vsn" => "2.0.0"}, params)) + url = "ws://127.0.0.1:#{@port}#{path}/websocket?" <> query + {:ok, client} = WebsocketClient.connect(self(), url, V2.JSONSerializer) + client + end + + defp control(client, event, payload) do + json = Phoenix.json_library().encode!([nil, nil, "phoenix", event, payload]) + :ok = WebsocketClient.send(client, {:text, json}) + end + + # connects with a new session and returns its token + defp new_session(params \\ %{}) do + client = connect(Map.put(params, "resumable", "1")) + assert_receive %Message{topic: "phoenix", event: "phx_session", payload: %{"token" => token}} + {client, token} + end + + defp join(client, topic) do + WebsocketClient.join(client, topic, %{}) + assert_receive %Message{event: "phx_reply", topic: ^topic, payload: %{"status" => "ok"}} + assert_receive %Message{event: "joined", topic: ^topic} + Process.whereis(String.to_atom(topic)) + end + + # the client resumes over a new connection with the join ref of the first one + defp echo(client, payload) do + json = Phoenix.json_library().encode!(["11", "99", "room:a", "echo", payload]) + :ok = WebsocketClient.send(client, {:text, json}) + end + + # drops the connection without closing it, like a network failure + defp drop(client) do + ref = Process.monitor(client) + Process.unlink(client) + Process.exit(client, :kill) + assert_receive {:DOWN, ^ref, _, _, _} + end + + defp resume(token, ack, params \\ %{}, path \\ "/ws") do + client = connect(Map.merge(params, %{"resumable" => "1", "resume" => "1"}), path) + control(client, "phx_resume", %{token: token, ack: ack}) + client + end + + defp poll(method, token, params, body \\ nil) do + query = URI.encode_query(Map.merge(%{"vsn" => "2.0.0"}, params)) + url = "http://127.0.0.1:#{@port}/ws/longpoll?" <> query + + headers = + if token, + do: %{"x-phoenix-longpoll-token" => token, "content-type" => "application/x-ndjson"}, + else: %{} + + {:ok, resp} = HTTPClient.request(method, url, headers, body || "") + Phoenix.json_library().decode!(resp.body) + end + + defp decode(messages) do + for message <- messages, do: V2.JSONSerializer.decode!(message, opcode: :text) + end + + # a token with the pid of the session, as when another session gets the pid + # of a session that stopped + defp forge_token(token, fun) do + {:ok, {:v1, endpoint_id, session, priv_topic}} = + Phoenix.Token.verify(Endpoint, Atom.to_string(__MODULE__), token) + + data = fun.({:v1, endpoint_id, session, priv_topic}) + Phoenix.Transports.Session.sign_token(Endpoint, data, crypto: [max_age: 1_209_600]) + end + + defp session(token) do + {:ok, {:v1, _id, session, _topic}} = + Phoenix.Token.verify(Endpoint, Atom.to_string(__MODULE__), token) + + session + end + + for adapter <- [Bandit.PhoenixAdapter, Phoenix.Endpoint.Cowboy2Adapter] do + describe "adapter: #{inspect(adapter)}" do + @describetag adapter: adapter + + test "starts a session that the client resumes after losing its connection" do + {client, token} = new_session() + assert_received {:connect, _} + channel = join(client, "room:a") + # the client received the join reply and the joined push + drop(client) + + # the channel keeps running and its messages are kept for the client + send(channel, {:push, "missed", %{n: 1}}) + send(channel, {:push, "missed", %{n: 2}}) + + client = resume(token, 2) + assert_receive {:resume, %{"vsn" => "2.0.0"} = params} + refute Map.has_key?(params, "resume") + assert_receive %Message{event: "phx_resumed", payload: %{"seq" => 1}} + assert_receive %Message{event: "missed", payload: %{"n" => 1}} + assert_receive %Message{event: "missed", payload: %{"n" => 2}} + assert Process.whereis(:"room:a") == channel + + # the channel keeps working + echo(client, %{"hi" => true}) + assert_receive %Message{event: "echo", payload: %{"hi" => true}} + refute_received {:connect, _} + end + + test "sends the messages the client did not receive again" do + {client, token} = new_session() + _channel = join(client, "room:a") + drop(client) + + # the client lost the joined push + _client = resume(token, 1) + assert_receive %Message{event: "phx_resumed"} + assert_receive %Message{event: "joined"} + end + + test "tells the client about the messages the session received" do + {client, _token} = new_session() + join(client, "room:a") + assert_receive %Message{event: "phx_ack", payload: %{"seq" => 1}} + + WebsocketClient.send_event(client, "room:a", "echo", %{}) + assert_receive %Message{event: "phx_ack", payload: %{"seq" => 2}} + end + + test "drops the messages the client acknowledged" do + {client, token} = new_session() + join(client, "room:a") + session = session(token) + assert :queue.len(:sys.get_state(session).out) == 2 + + control(client, "phx_ack", %{ack: 2}) + Process.sleep(50) + assert :queue.is_empty(:sys.get_state(session).out) + end + + test "drops the messages of the previous connection the client sends again" do + {client, token} = new_session() + join(client, "room:a") + session = session(token) + drop(client) + + # the old connection delivers the echo late, after the client resumed + client = resume(token, 2) + assert_receive %Message{event: "phx_resumed", payload: %{"seq" => 1}} + frame = ~s(["11","9","room:a","echo",{"n":1}]) + send(session, {:pipe_in, self(), 2, {frame, :text}}) + assert_receive %Message{event: "echo", payload: %{"n" => 1}} + + # the client sends it again + WebsocketClient.send(client, {:text, frame}) + echo(client, %{"n" => 2}) + assert_receive %Message{event: "echo", payload: %{"n" => 2}} + refute_received %Message{event: "echo", payload: %{"n" => 1}} + end + + test "stops the session when it rejects the resume" do + {client, token} = new_session() + channel = join(client, "room:a") + ref = Process.monitor(channel) + drop(client) + + resume(token, 2, %{"reject_resume" => "true"}) + assert_receive %Message{event: "phx_gone", payload: payload} + refute Map.has_key?(payload, "code") + assert_receive {:DOWN, ^ref, _, _, _} + end + + test "tells the client when it cannot resume the session" do + {client, token} = new_session() + session = session(token) + ref = Process.monitor(session) + drop(client) + + # the session expires without a connection + assert_receive {:DOWN, ^ref, _, _, {:shutdown, :inactive}}, 1000 + + resume(token, 0) + assert_receive %Message{event: "phx_gone"} + + resume("invalid", 0) + assert_receive %Message{event: "phx_gone"} + end + + test "requires the client to resume before sending messages" do + client = connect(%{"resumable" => "1", "resume" => "1"}) + WebsocketClient.join(client, "room:a", %{}) + assert_receive %Message{event: "phx_gone"} + refute_received {:connect, _} + end + + test "replaces the connection the client gave up on" do + {client, token} = new_session() + join(client, "room:a") + ref = Process.monitor(client) + + # the client resumes while the server still has the old connection + _client = resume(token, 2) + assert_receive %Message{event: "phx_resumed"} + assert_receive {:DOWN, ^ref, _, _, _} + end + + test "stops the session when the client closes it" do + {client, _token} = new_session() + channel = join(client, "room:a") + ref = Process.monitor(channel) + + control(client, "phx_close", %{}) + assert_receive {:DOWN, ^ref, _, _, {:shutdown, :closed}} + end + + test "tells the client when the session stops" do + {client, _token} = new_session(%{"user_id" => "1"}) + join(client, "room:a") + + Endpoint.broadcast("user_sockets:1", "disconnect", %{}) + assert_receive %Message{event: "phx_gone", payload: %{"code" => 1011}} + end + + test "closes the session when the client does not acknowledge messages" do + {client, token} = new_session() + join(client, "room:a") + ref = Process.monitor(session(token)) + + assert_receive {:DOWN, ^ref, _, _, {:shutdown, :unacknowledged}}, 1000 + assert_receive %Message{event: "phx_gone"} + end + + test "long polling resumes a WebSocket session" do + {client, token} = new_session() + channel = join(client, "room:a") + drop(client) + send(channel, {:push, "missed", %{}}) + + body = poll(:get, token, %{"ack" => "2", "resume" => "1"}) + assert body["status"] == 200 + assert [%Message{event: "missed"}] = decode(body["messages"]) + assert body["last"] == 3 + assert_receive {:resume, _params} + + assert {_, %Phoenix.Socket{transport: :longpoll}} = + elem(:sys.get_state(session(token)).handler, 1) + + # the client sends the messages the WebSocket did not deliver, numbered + # after the last one the session confirmed + frame = ~s(["11","9","room:a","echo",{"n":1}]) + assert poll(:post, token, %{"seq" => "2"}, frame)["status"] == 200 + assert poll(:post, token, %{"seq" => "2"}, frame)["status"] == 200 + + body = poll(:get, token, %{"ack" => "3"}) + assert [%Message{event: "echo", payload: %{"n" => 1}}] = decode(body["messages"]) + assert Process.whereis(:"room:a") == channel + end + + test "long polling takes over from a WebSocket the client gave up on" do + {client, token} = new_session() + join(client, "room:a") + ref = Process.monitor(client) + + body = poll(:get, token, %{"ack" => "2", "resume" => "1"}) + assert body["status"] == 204 + assert_receive {:DOWN, ^ref, _, _, _} + assert :sys.get_state(session(token)).pipe == nil + end + + test "polls do not take the messages of a WebSocket" do + {client, token} = new_session() + channel = join(client, "room:a") + + # a poll the client gave up on when it resumed over the WebSocket + task = Task.async(fn -> poll(:get, token, %{"ack" => "0"}) end) + Process.sleep(50) + send(channel, {:push, "new", %{}}) + assert_receive %Message{event: "new"} + assert Task.await(task)["status"] == 204 + end + + test "long polling sends binary messages in base64" do + {client, token} = new_session() + join(client, "room:a") + drop(client) + + assert poll(:post, token, %{"seq" => "2"}, ~s(["11","9","room:a","binary",{}])) + body = poll(:get, token, %{"ack" => "2", "resume" => "1"}) + assert ["base64:" <> encoded] = body["messages"] + + assert <<0, _join_ref_size, _topic_size, _event_size, rest::binary>> = + Base.decode64!(encoded) + + assert String.ends_with?(rest, <<1, 2, 3>>) + end + + test "a WebSocket resumes a long polling session" do + body = poll(:get, nil, %{}) + assert body["status"] == 410 + token = body["token"] + + join = ~s(["11","1","room:a","phx_join",{}]) + assert poll(:post, token, %{"seq" => "1"}, join)["status"] == 200 + channel = Process.whereis(:"room:a") + Process.sleep(50) + + # the client did not receive the join reply and the joined push + resume(token, 0) + assert_receive %Message{event: "phx_resumed", payload: %{"seq" => 1}} + assert_receive %Message{event: "phx_reply", topic: "room:a"} + assert_receive %Message{event: "joined", topic: "room:a"} + assert Process.whereis(:"room:a") == channel + end + + test "a WebSocket does not resume the session of another token with the same pid" do + {client, token} = new_session() + channel = join(client, "room:a") + session = session(token) + pipe = :sys.get_state(session).pipe + forged = forge_token(token, fn {:v1, id, pid, _topic} -> {:v1, id, pid, "other"} end) + + _other = resume(forged, 0) + assert_receive %Message{event: "phx_gone"} + refute_received {:resume, _} + + # the session keeps its client + assert :sys.get_state(session).pipe == pipe + echo(client, %{"still" => "here"}) + assert_receive %Message{event: "echo", payload: %{"still" => "here"}} + assert Process.whereis(:"room:a") == channel + end + + test "a WebSocket does not resume a session of an endpoint that restarted" do + {_client, token} = new_session() + forged = forge_token(token, fn {:v1, _id, pid, topic} -> {:v1, "old", pid, topic} end) + + resume(forged, 0) + assert_receive %Message{event: "phx_gone"} + refute_received {:resume, _} + end + + test "does not start a session for clients that don't resume" do + client = connect(%{}) + join(client, "room:a") + refute_received %Message{event: "phx_session"} + end + + test "does not start a session when the socket is not resumable" do + client = connect(%{"resumable" => "1"}, "/ws/plain") + assert_receive {:connect, %{"vsn" => "2.0.0"}} + join(client, "room:a") + refute_received %Message{event: "phx_session"} + end + + test "does not pass the resume params to the socket" do + new_session() + assert_receive {:connect, params} + refute Map.has_key?(params, "resumable") + end + + test "starts a session when the :resumable function allows the connection" do + client = connect(%{"resumable" => "1", "user_id" => "1"}, "/ws/users") + assert_receive %Message{event: "phx_session", payload: %{"token" => _}} + + # the function receives the same map as connect/1, followed by its arguments + assert_received {:connect, params} + assert_received {:resumable?, transport_info, :arg} + assert %{endpoint: Endpoint, transport: :websocket, connect_info: %{}} = transport_info + assert transport_info.params == params + refute Map.has_key?(params, "resumable") + + join(client, "room:a") + end + + test "does not start a session when the :resumable function denies the connection" do + client = connect(%{"resumable" => "1"}, "/ws/users") + assert_received {:connect, %{"vsn" => "2.0.0"}} + assert_received {:resumable?, _transport_info, :arg} + + # the client connects like it does to a socket that is not resumable + join(client, "room:a") + refute_received %Message{event: "phx_session"} + end + + test "does not invoke the :resumable function when a client resumes" do + client = connect(%{"resumable" => "1", "user_id" => "1"}, "/ws/users") + assert_receive %Message{event: "phx_session", payload: %{"token" => token}} + assert_received {:resumable?, _transport_info, :arg} + channel = join(client, "room:a") + drop(client) + + # the function denies connections without a user + resume(token, 2, %{}, "/ws/users") + assert_receive %Message{event: "phx_resumed"} + assert_received {:resume, _params} + refute_received {:resumable?, _transport_info, _arg} + assert Process.whereis(:"room:a") == channel + end + + test "does not invoke the :resumable function when the socket rejects the connection" do + params = %{"vsn" => "2.0.0", "resumable" => "1", "reject_connect" => "true"} + url = "ws://127.0.0.1:#{@port}/ws/users/websocket?" <> URI.encode_query(params) + + assert {:error, %Mint.WebSocket.UpgradeFailureError{status_code: 403}} = + WebsocketClient.connect(self(), url, V2.JSONSerializer) + + refute_received {:resumable?, _transport_info, _arg} + end + + test "raises when the :resumable function does not return a boolean" do + # we call the endpoint ourselves, as the server logs the error after it responded + params = %{"vsn" => "2.0.0", "resumable" => "1", "user_id" => "invalid"} + conn = Plug.Test.conn(:get, "/ws/users/websocket?" <> URI.encode_query(params)) + + assert_raise ArgumentError, ~r/resumable\?\/2, given as :resumable, must return/, fn -> + Endpoint.call(conn, []) + end + end + end + end +end diff --git a/test/support/websocket_client.exs b/test/support/websocket_client.exs index ab001975aa..d66354f227 100644 --- a/test/support/websocket_client.exs +++ b/test/support/websocket_client.exs @@ -16,6 +16,8 @@ defmodule Phoenix.Integration.WebsocketClient do :sender, :serializer, closing?: false, + # frames that arrive together with the upgrade response, before we have the websocket + early_data: [], topics: %{}, # Use different initial join_ref from ref to # make sure the server is not coupling them. @@ -175,9 +177,12 @@ defmodule Phoenix.Integration.WebsocketClient do defp handle_responses(%{request_ref: ref} = state, [{:done, ref} | rest]) do case Mint.WebSocket.new(state.conn, ref, state.status, state.resp_headers) do {:ok, conn, websocket} -> + early_data = for data <- Enum.reverse(state.early_data), do: {:data, ref, data} + %{state | conn: conn, websocket: websocket, status: nil, resp_headers: nil} + |> Map.put(:early_data, []) |> reply({:ok, :connected}) - |> handle_responses(rest) + |> handle_responses(early_data ++ rest) {:error, conn, reason} -> put_in(state.conn, conn) @@ -201,6 +206,10 @@ defmodule Phoenix.Integration.WebsocketClient do end end + defp handle_responses(%{request_ref: ref} = state, [{:data, ref, data} | rest]) do + handle_responses(%{state | early_data: [data | state.early_data]}, rest) + end + defp handle_responses(state, [_response | rest]) do handle_responses(state, rest) end