diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index e1918c5..2ef1a17 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -1,9 +1,9 @@ -# This workflow will run tests using node and then publish a package to GitHub Packages when a release is created -# For more information see: https://docs.github.com/en/actions/publishing-packages/publishing-nodejs-packages +# Run tests for pull requests, releases, and manual dispatches. Publish only for releases and manual dispatches. name: Node.js Package on: + pull_request: release: types: [created] workflow_dispatch: @@ -20,6 +20,7 @@ jobs: - run: npm test publish-npm: + if: github.event_name == 'release' || github.event_name == 'workflow_dispatch' needs: build runs-on: ubuntu-latest steps: diff --git a/connection.js b/connection.js index 0b5ea45..87b55cf 100644 --- a/connection.js +++ b/connection.js @@ -107,6 +107,12 @@ class DiodeResponseError extends DiodeConnectionError { } } +// Legacy relays reject a ticket more than 100,000,000 bytes above the last +// accepted ticket. Current relays compare against device usage instead. +const MAX_TICKET_BYTE_ADVANCE = 64_000_000; +const TICKET_HEADROOM_LEVELS = [1024, 64 * 1024, 256 * 1024, 1024 * 1024]; +const RELAY_USAGE_HELLO_VERSION = 1001; + class DiodeConnection extends EventEmitter { constructor(host, port, keyLocation = './db/keys.json') { super(); @@ -190,7 +196,16 @@ class DiodeConnection extends EventEmitter { this._ticketUpdateToken = null; this.pendingTicketUpdateForce = false; this.lastAcceptedTicketBytes = 0; + this.lastAcceptedTicketEpoch = null; this.lastRelayMeasuredBytes = 0; + this._relayUsage = null; + this._relayUsageGeneration = -1; + this._relayUsageSequence = 0; + this._supportsRelayUsage = null; + this._relayUsageQueryInFlight = false; + this._ticketAcceptedOnTransport = false; + this._ticketHeadroomBytes = TICKET_HEADROOM_LEVELS[0]; + this.relayUsageWaitMs = this.commandTimeoutMs; // Log the ticket batching settings logger.info(() => `Ticket batching settings - Bytes Threshold: ${this.ticketUpdateThreshold} bytes, Update Interval: ${this.ticketUpdateInterval}ms`); @@ -254,6 +269,19 @@ class DiodeConnection extends EventEmitter { this._transportReady = false; this.receiveBuffer = Buffer.alloc(0); this._serverEthAddress = null; + // A new TLS connection can terminate on a different relay with a + // different paid floor. Never carry a local high-water ticket into it. + this.totalBytes = 128000; + this.accumulatedBytes = 0; + this.lastAcceptedTicketBytes = 0; + this.lastAcceptedTicketEpoch = null; + this.lastRelayMeasuredBytes = 0; + this._relayUsage = null; + this._relayUsageGeneration = -1; + this._supportsRelayUsage = null; + this._relayUsageQueryInFlight = false; + this._ticketAcceptedOnTransport = false; + this._ticketHeadroomBytes = TICKET_HEADROOM_LEVELS[0]; const attempt = new Promise((resolve, reject) => { let settled = false; @@ -310,9 +338,9 @@ class DiodeConnection extends EventEmitter { const cachedServerAddress = await this._waitForServerEthereumAddress(handshakeOptions); this._assertCurrentTransport(socket, generation, 'Diode server identity lookup'); if (cachedServerAddress) this._serverEthAddress = cachedServerAddress; - this._assertCurrentTransport(socket, generation, 'Diode measured-byte synchronization'); - await this._syncMeasuredBytesWithRelay(handshakeOptions); - this._assertCurrentTransport(socket, generation, 'Diode measured-byte synchronization'); + this._assertCurrentTransport(socket, generation, 'Diode relay usage synchronization'); + await this._refreshTicketUsage(handshakeOptions); + this._assertCurrentTransport(socket, generation, 'Diode relay usage synchronization'); const ticketCommand = await this.createTicketCommand(handshakeOptions); this._assertCurrentTransport(socket, generation, 'Diode ticket creation'); const ticketResponse = await this._sendCommandTransportReady(ticketCommand, handshakeOptions); @@ -332,6 +360,17 @@ class DiodeConnection extends EventEmitter { this.isReconnecting = false; this.retryCount = 0; this._startTicketUpdateTimer(); + // A bounded handshake ticket can leave a large unpaid backlog. + // Continue paying it promptly instead of waiting for the timer or + // another packet to arrive. + if (this.accumulatedBytes >= this.ticketUpdateThreshold) { + setImmediate(() => { + if (generation !== this._socketGeneration || !this.isReady()) return; + this._updateTicketIfNeeded(true).catch((error) => { + logger.error(() => `Error catching up connection tickets: ${error}`); + }); + }); + } if (this._resolveReconnectWaiter) this._resolveReconnectWaiter(); settle(); } catch (error) { @@ -752,19 +791,20 @@ class DiodeConnection extends EventEmitter { const originalCommand = pending.commandArray; const retryCount = pending.ticketRetryCount || 0; const isTicketCommand = this._isTicketCommand(originalCommand); + const maxTicketRetries = this._supportsRelayUsage === true + ? TICKET_HEADROOM_LEVELS.length - 1 + : 1; - if (isTicketCommand && retryCount < 1) { + if (isTicketCommand && retryCount < maxTicketRetries) { this.fixResponse(responseData); + if (this._supportsRelayUsage === true) { + this._ticketHeadroomBytes = TICKET_HEADROOM_LEVELS[retryCount + 1]; + } const allowTransportReady = pending.allowTransportReady === true; const expectedSocket = pending.expectedSocket; const expectedGeneration = pending.expectedGeneration; - const hasExpectedTransport = expectedSocket !== undefined || expectedGeneration !== undefined; const retryContext = { allowTransportReady, expectedSocket, expectedGeneration }; - this._syncMeasuredBytesWithRelay(retryContext) - .catch((error) => { - if (hasExpectedTransport) throw error; - logger.debug(() => `Unable to sync relay measured bytes after too_low: ${error}`); - }) + this._refreshTicketUsage(retryContext) .then(() => this.createTicketCommand(retryContext)) .then((ticketCommand) => { const retryOptions = { @@ -779,7 +819,10 @@ class DiodeConnection extends EventEmitter { : this.sendCommand(ticketCommand, retryOptions); }) .then(pending.resolve) - .catch(pending.reject); + .catch((error) => { + this._ticketHeadroomBytes = TICKET_HEADROOM_LEVELS[0]; + pending.reject(error); + }); return true; } this._recordTicketResponse(originalCommand, responseData); @@ -882,14 +925,23 @@ class DiodeConnection extends EventEmitter { if (messageType === 'ticket_request') { const deviceUsageRaw = messageContent[1]; const deviceUsage = parseUInt(deviceUsageRaw); - if (typeof deviceUsage === 'number' && deviceUsage > this.totalBytes) { - this.totalBytes = deviceUsage; + if (Number.isSafeInteger(deviceUsage) && deviceUsage >= 0) { + this._relayUsage = deviceUsage; + this._relayUsageGeneration = this._socketGeneration; + this._relayUsageSequence += 1; + this.emit('relay_usage', { + generation: this._socketGeneration, + sequence: this._relayUsageSequence, + usage: deviceUsage, + }); } - // Use the same single-flight ticket path as periodic updates. - this._updateTicketIfNeeded(true).catch((error) => { - logger.error(() => `Error handling ticket_request: ${error}`); - }); + if (this.isReady() && !this._relayUsageQueryInFlight) { + // Use the same single-flight ticket path as periodic updates. + this._updateTicketIfNeeded(true).catch((error) => { + logger.error(() => `Error handling ticket_request: ${error}`); + }); + } } return true; } @@ -906,19 +958,28 @@ class DiodeConnection extends EventEmitter { local_address, device_signature ] - The byte value is the relay's last accepted paid floor. The live - unpaid measurement still comes from the relay "bytes" command. + The byte value is the relay's last accepted paid floor. It says + nothing about new usage; the relay usage report supplies that. */ const lastTicket = this._parseTooLowTicketSummary(response); + if (Number.isSafeInteger(lastTicket.epoch) && + Number.isSafeInteger(this.lastAcceptedTicketEpoch) && + lastTicket.epoch < this.lastAcceptedTicketEpoch) return; if (Number.isFinite(lastTicket.totalConnections)) { this.totalConnections = Math.max(this.totalConnections, lastTicket.totalConnections); } if (Number.isFinite(lastTicket.totalBytes)) { - this.lastAcceptedTicketBytes = Math.max(this.lastAcceptedTicketBytes, lastTicket.totalBytes); - const relayMeasuredBytes = Number.isFinite(this.lastRelayMeasuredBytes) ? this.lastRelayMeasuredBytes : 0; - const pendingBytes = Math.max(this.accumulatedBytes, relayMeasuredBytes, 0); - this.totalBytes = Math.max(this.totalBytes, lastTicket.totalBytes + pendingBytes + 1024); - this.accumulatedBytes = Math.max(this.accumulatedBytes, this.totalBytes - lastTicket.totalBytes); + if (Number.isSafeInteger(lastTicket.epoch) && lastTicket.epoch !== this.lastAcceptedTicketEpoch) { + this.lastAcceptedTicketEpoch = lastTicket.epoch; + this.lastAcceptedTicketBytes = lastTicket.totalBytes; + } else { + this.lastAcceptedTicketBytes = Math.max(this.lastAcceptedTicketBytes, lastTicket.totalBytes); + } + // Local pending bytes can already be included in the relay's global + // usage and another connection's paid ticket. A fresh relay query after + // too_low establishes the target without adding them again. + this.totalBytes = this.lastAcceptedTicketBytes; + this.accumulatedBytes = 0; } } @@ -982,10 +1043,25 @@ class DiodeConnection extends EventEmitter { _recordTicketResponse(commandArray, responseData) { if (!this._isTicketCommand(commandArray) || !Array.isArray(responseData)) return; const status = responseData[0] !== undefined ? parseResponseType(responseData[0]) : ''; - if (status !== 'thanks!') return; + if (status !== 'thanks!') { + this._ticketHeadroomBytes = TICKET_HEADROOM_LEVELS[0]; + return; + } const ticketTotalBytes = this._ticketTotalBytes(commandArray); if (!Number.isFinite(ticketTotalBytes)) return; + if (commandArray[0] === 'ticketv2') { + const ticketEpoch = parseUInt(commandArray[2]); + if (Number.isSafeInteger(ticketEpoch) && + Number.isSafeInteger(this.lastAcceptedTicketEpoch) && + ticketEpoch < this.lastAcceptedTicketEpoch) return; + if (Number.isSafeInteger(ticketEpoch) && ticketEpoch !== this.lastAcceptedTicketEpoch) { + this.lastAcceptedTicketEpoch = ticketEpoch; + this.lastAcceptedTicketBytes = 0; + } + } this.lastAcceptedTicketBytes = Math.max(this.lastAcceptedTicketBytes, ticketTotalBytes); + this._ticketAcceptedOnTransport = true; + this._ticketHeadroomBytes = TICKET_HEADROOM_LEVELS[0]; this.accumulatedBytes = Math.max(0, this.totalBytes - ticketTotalBytes); this.lastTicketUpdate = Date.now(); } @@ -1007,6 +1083,116 @@ class DiodeConnection extends EventEmitter { return -((encoded - 1) / 2); } + _waitForRelayUsageAfter(sequence, generation) { + if (generation !== this._socketGeneration || generation === this._lastDisconnectedGeneration) { + return Promise.reject(new DiodeDisconnectedError('Diode relay disconnected before reporting usage')); + } + if (this._relayUsageGeneration === generation && this._relayUsageSequence > sequence) { + return Promise.resolve(this._relayUsage); + } + return new Promise((resolve, reject) => { + let timer; + let settleImmediate; + let latestUsage; + const cleanup = () => { + clearTimeout(timer); + if (settleImmediate) clearImmediate(settleImmediate); + this.off('relay_usage', onUsage); + this.off('disconnect', onDisconnect); + }; + const onUsage = (event) => { + if (event.generation !== generation || event.sequence <= sequence) return; + latestUsage = event.usage; + if (settleImmediate) return; + settleImmediate = setImmediate(() => { + cleanup(); + resolve(latestUsage); + }); + }; + const onDisconnect = (event) => { + if (event.generation !== generation) return; + cleanup(); + reject(new DiodeDisconnectedError('Diode relay disconnected before reporting usage')); + }; + this.on('relay_usage', onUsage); + this.on('disconnect', onDisconnect); + timer = setTimeout(() => { + cleanup(); + reject(new DiodeConnectionError('Diode relay did not report ticket usage', 'DIODE_USAGE_TIMEOUT')); + }, normalizeTimerMs(this.relayUsageWaitMs, this.commandTimeoutMs)); + }); + } + + async _refreshTicketUsage({ + allowTransportReady = false, + expectedSocket = undefined, + expectedGeneration = undefined, + } = {}) { + const hasExpectedTransport = expectedSocket !== undefined || expectedGeneration !== undefined; + const generation = hasExpectedTransport ? expectedGeneration : this._socketGeneration; + const commandOptions = hasExpectedTransport ? { expectedSocket, expectedGeneration } : {}; + const send = allowTransportReady + ? (command) => this._sendCommandTransportReady(command, commandOptions) + : (command) => this.sendCommand(command, commandOptions); + + // The relay's paid ticket is scoped to an epoch; never use the previous + // epoch's floor when interpreting current device usage. + const epoch = allowTransportReady + ? await this.RPC._getEpochWithSender(send) + : await this.RPC.getEpoch(); + if (hasExpectedTransport) this._assertCurrentTransport(expectedSocket, expectedGeneration, 'Diode usage epoch'); + if (epoch !== this.lastAcceptedTicketEpoch) { + this.lastAcceptedTicketEpoch = epoch; + this.lastAcceptedTicketBytes = 0; + } + + if (this._supportsRelayUsage === false) { + if (!this._ticketAcceptedOnTransport) { + // Older relays do not provide an absolute usage floor. Probe with a + // small ticket; too_low supplies their paid floor. Never add an + // ambiguous pre-ticket `bytes` reply to that floor. + this.totalBytes = Math.max(128000, this.lastAcceptedTicketBytes); + this.accumulatedBytes = Math.max(0, this.totalBytes - this.lastAcceptedTicketBytes); + return null; + } + return this._syncMeasuredBytesWithRelay({ allowTransportReady, expectedSocket, expectedGeneration }); + } + + const sequence = this._relayUsageSequence; + this._relayUsageQueryInFlight = true; + try { + let response; + try { + response = await send(['hello', RELAY_USAGE_HELLO_VERSION]); + } catch (error) { + if (!/version not supported/i.test(String(error && error.message))) throw error; + this._supportsRelayUsage = false; + return this._refreshTicketUsage({ allowTransportReady, expectedSocket, expectedGeneration }); + } + if (!response || parseResponseType(response[0]) !== 'ok') { + throw new DiodeConnectionError('Diode relay rejected usage negotiation', 'DIODE_USAGE_NEGOTIATION'); + } + if (hasExpectedTransport) this._assertCurrentTransport(expectedSocket, expectedGeneration, 'Diode usage negotiation'); + // The relay sends its fresh ticket_request before the hello response, + // but unsolicited dispatch is deferred. Drain that batch so an older + // queued request cannot win over the fresh report. + await new Promise(setImmediate); + const usage = await this._waitForRelayUsageAfter(sequence, generation); + if (hasExpectedTransport) this._assertCurrentTransport(expectedSocket, expectedGeneration, 'Diode usage report'); + if (!Number.isSafeInteger(usage) || usage < 0) { + throw new DiodeConnectionError('Diode relay reported invalid usage', 'DIODE_USAGE_INVALID'); + } + this._supportsRelayUsage = true; + this.totalBytes = this.lastAcceptedTicketBytes > 0 && usage <= this.lastAcceptedTicketBytes + ? this.lastAcceptedTicketBytes + : usage + this._ticketHeadroomBytes; + this.accumulatedBytes = Math.max(0, this.totalBytes - this.lastAcceptedTicketBytes); + return usage; + } finally { + this._relayUsageQueryInFlight = false; + } + } + async _syncMeasuredBytesWithRelay({ allowTransportReady = false, expectedSocket = undefined, @@ -1027,19 +1213,19 @@ class DiodeConnection extends EventEmitter { const measuredBytes = responseData && responseData[0] !== undefined ? this._parseRelaySignedInt(responseData[0]) : null; - if (!Number.isFinite(measuredBytes) || measuredBytes <= 0) { - return measuredBytes; + if (!Number.isSafeInteger(measuredBytes)) { + throw new DiodeConnectionError('Diode relay returned invalid unpaid bytes', 'DIODE_USAGE_INVALID'); } this.lastRelayMeasuredBytes = measuredBytes; const baseBytes = Number.isFinite(this.lastAcceptedTicketBytes) && this.lastAcceptedTicketBytes > 0 ? this.lastAcceptedTicketBytes : 128000; - const targetTotalBytes = baseBytes + measuredBytes + 1024; - if (targetTotalBytes > this.totalBytes) { - this.totalBytes = targetTotalBytes; - this.accumulatedBytes = Math.max(this.accumulatedBytes, this.totalBytes - baseBytes); - } + // On old relays this command is used only after an accepted ticket has + // selected the right fleet. Reconcile downward as well as upward: writes + // lost on disconnect must not become payable traffic. + this.totalBytes = measuredBytes <= 0 ? baseBytes : baseBytes + measuredBytes; + this.accumulatedBytes = Math.max(0, this.totalBytes - baseBytes); return measuredBytes; } @@ -1419,9 +1605,6 @@ class DiodeConnection extends EventEmitter { const totalConnections = this.totalConnections + 1; if (!hasExpectedTransport) this.totalConnections = totalConnections; - // Assume totalBytes is managed elsewhere - const totalBytes = this.totalBytes; - // Get server Ethereum address as Buffer const serverIdBuffer = await this._waitForServerEthereumAddress({ expectedSocket, @@ -1444,6 +1627,28 @@ class DiodeConnection extends EventEmitter { if (hasExpectedTransport) { this._assertCurrentTransport(expectedSocket, expectedGeneration, 'Diode ticket epoch'); } + // Relay ticket floors are scoped to an epoch. The previous epoch's + // accepted total cannot bound a ticket in the new epoch. + if (epoch !== this.lastAcceptedTicketEpoch) { + this.lastAcceptedTicketEpoch = epoch; + this.lastAcceptedTicketBytes = 0; + } + // Preserve the full unpaid target in this.totalBytes. Only bound the + // amount signed into this ticket; _recordTicketResponse advances the + // accepted floor after each acknowledgement. + const acceptedBytes = Number.isSafeInteger(this.lastAcceptedTicketBytes) + ? Math.max(0, this.lastAcceptedTicketBytes) + : 0; + let totalBytes; + if (this._supportsRelayUsage === true && this._relayUsageGeneration === this._socketGeneration) { + totalBytes = acceptedBytes > 0 && this._relayUsage <= acceptedBytes + ? acceptedBytes + : this._relayUsage + this._ticketHeadroomBytes; + this.totalBytes = totalBytes; + this.accumulatedBytes = Math.max(0, totalBytes - acceptedBytes); + } else { + totalBytes = Math.min(this.totalBytes, acceptedBytes + MAX_TICKET_BYTE_ADVANCE); + } const signature = await this.createTicketSignature( serverIdBuffer, totalConnections, @@ -1567,9 +1772,7 @@ class DiodeConnection extends EventEmitter { try { if (this.accumulatedBytes > 0 || force) { logger.debug(() => `Updating ticket: accumulated ${this.accumulatedBytes} bytes, ${timeSinceLastUpdate}ms since last update`); - await this._syncMeasuredBytesWithRelay().catch((error) => { - logger.debug(() => `Unable to sync relay measured bytes before ticket update: ${error}`); - }); + await this._refreshTicketUsage(); if (generation !== this._socketGeneration || this._ticketUpdateToken !== updateToken) return; const ticketCommand = await this.createTicketCommand(); if (generation !== this._socketGeneration || this._ticketUpdateToken !== updateToken) return; diff --git a/package-lock.json b/package-lock.json index 050ebdb..399f9c2 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "diodejs", - "version": "0.5.8", + "version": "0.5.9", "lockfileVersion": 2, "requires": true, "packages": { "": { "name": "diodejs", - "version": "0.5.8", + "version": "0.5.9", "license": "ISC", "dependencies": { "@ethereumjs/rlp": "^5.0.2", diff --git a/package.json b/package.json index d34973b..14a1b53 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "diodejs", - "version": "0.5.8", + "version": "0.5.9", "description": "A JavaScript client for interacting with the Diode network. It provides functionalities to bind and publish ports, send RPC commands, and handle responses.", "main": "index.js", "files": [ diff --git a/test/connection.stability.test.js b/test/connection.stability.test.js index 535985b..d67e77d 100644 --- a/test/connection.stability.test.js +++ b/test/connection.stability.test.js @@ -181,6 +181,14 @@ test('malformed deferred unsolicited frames cannot escape or block a later ticke await new Promise((resolve) => setImmediate(resolve)); assert.equal(applicationListenerCalls, malformedFrames.length + 1); + assert.equal(connection._relayUsage, 131072); + assert.equal(ticketRequests, 0, 'a ticket request cannot start before the handshake'); + prepareTransport(connection); + connection._deferUnsolicited([ + Buffer.from([4]), + [Buffer.from('ticket_request'), Buffer.from([2, 0, 1])], + ]); + await new Promise((resolve) => setImmediate(resolve)); assert.equal(ticketRequests, 1); connection.close(); }); @@ -195,7 +203,7 @@ test('too_low response does not leave processed frames to be replayed', async () resolve, reject, commandArray: ['ticketv2'], - ticketRetryCount: 1, + ticketRetryCount: 3, }); }); const tooLow = [ @@ -283,7 +291,7 @@ test('an old ticket update cannot clear a new session ticket flight', async () = const first = deferred(); const second = deferred(); let syncCall = 0; - connection._syncMeasuredBytesWithRelay = () => { + connection._refreshTicketUsage = () => { syncCall += 1; return syncCall === 1 ? first.promise : second.promise; }; @@ -421,7 +429,7 @@ test('connection becomes ready only after the secure ticket handshake succeeds', const connection = makeConnection(); connection._waitForServerEthereumAddress = async () => Buffer.alloc(20, 1); - connection._syncMeasuredBytesWithRelay = async () => 0; + connection._refreshTicketUsage = async () => 0; connection.createTicketCommand = async () => ['ticketv2']; connection._sendCommandTransportReady = async () => ['thanks!']; try { @@ -462,7 +470,7 @@ test('a stale handshake cannot mutate or send through a newer ready generation', lookupCalls += 1; return lookupCalls === 1 ? firstLookup.promise : secondServerAddress; }; - connection._syncMeasuredBytesWithRelay = async () => { + connection._refreshTicketUsage = async () => { syncGenerations.push(connection._socketGeneration); return 0; }; @@ -529,7 +537,7 @@ test('explicit connect resolves a backoff waiter and a later disconnect can reco connection.retryDelay = 1000; connection.maxRetryDelay = 1000; connection._waitForServerEthereumAddress = async () => Buffer.alloc(20, 1); - connection._syncMeasuredBytesWithRelay = async () => 0; + connection._refreshTicketUsage = async () => 0; connection.createTicketCommand = async () => ['ticketv2']; connection._sendCommandTransportReady = async () => ['thanks!']; try { diff --git a/test/connection.ticketRetry.test.js b/test/connection.ticketRetry.test.js index c99c012..173c3ef 100644 --- a/test/connection.ticketRetry.test.js +++ b/test/connection.ticketRetry.test.js @@ -20,6 +20,7 @@ function encodeResponse(requestId, response) { } function tooLowResponse(overrides = {}) { + const epoch = overrides.epoch === undefined ? 687 : overrides.epoch; const totalConnections = overrides.totalConnections === undefined ? 11 : overrides.totalConnections; const totalBytes = overrides.totalBytes === undefined ? 135591 : overrides.totalBytes; @@ -27,7 +28,7 @@ function tooLowResponse(overrides = {}) { 'response', 'too_low', 1284, - 687, + epoch, totalConnections, totalBytes, Buffer.from([0]), @@ -35,11 +36,34 @@ function tooLowResponse(overrides = {}) { ]; } +function stubTicketSigning(connection, epoch = 687) { + connection._waitForServerEthereumAddress = async () => Buffer.alloc(20, 1); + connection.RPC.getEpoch = async () => epoch; + connection.createTicketSignature = async () => Buffer.alloc(65, 1); +} + +function reportRelayUsage(connection, usage) { + connection._handleData(encodeResponse(99, ['ticket_request', usage])); +} + +function stubHelloUsage(connection, usage, { beforeResponse = false } = {}) { + const commands = []; + connection.sendCommand = async (command) => { + commands.push(command[0]); + assert.deepEqual(command, ['hello', 1001]); + reportRelayUsage(connection, usage); + if (beforeResponse) await new Promise(setImmediate); + return ['ok']; + }; + return commands; +} + test('ticket too_low response retries once with repaired ticket', async () => { const connection = makeConnection(); let retryCount = 0; connection.fixResponse = () => {}; + connection._refreshTicketUsage = async () => {}; connection.createTicketCommand = async () => ['ticketv2']; connection.sendCommand = async (command, options) => { retryCount += 1; @@ -71,7 +95,7 @@ test('ticket too_low during initial handshake keeps the transport-ready retry pa let internalRetries = 0; connection.fixResponse = () => {}; - connection._syncMeasuredBytesWithRelay = async (options) => { + connection._refreshTicketUsage = async (options) => { syncAllowed = options.allowTransportReady === true; }; connection.createTicketCommand = async (options) => { @@ -105,9 +129,11 @@ test('ticket too_low during initial handshake keeps the transport-ready retry pa assert.equal(internalRetries, 1); }); -test('ticket too_low response does not retry recursively', async () => { +test('ticket too_low response stops after the bounded retries', async () => { const connection = makeConnection(); let retryCount = 0; + connection._supportsRelayUsage = true; + connection._ticketHeadroomBytes = 1024 * 1024; connection.fixResponse = () => {}; connection.createTicketCommand = async () => { @@ -120,7 +146,7 @@ test('ticket too_low response does not retry recursively', async () => { resolve, reject, commandArray: ['ticketv2'], - ticketRetryCount: 1, + ticketRetryCount: 3, }); }); @@ -129,6 +155,45 @@ test('ticket too_low response does not retry recursively', async () => { assert.equal(retryCount, 0); assert.equal(Buffer.from(result[0]).toString('utf8'), 'too_low'); + assert.equal(connection._ticketHeadroomBytes, 1024); +}); + +test('legacy relay stops after one too_low retry', async () => { + const connection = makeConnection(); + connection._supportsRelayUsage = false; + let refreshes = 0; + connection._refreshTicketUsage = async () => { refreshes += 1; }; + const result = new Promise((resolve, reject) => { + connection.pendingRequests.set(1, { + resolve, + reject, + commandArray: ['ticketv2'], + ticketRetryCount: 1, + }); + }); + + connection._handleData(encodeResponse(1, tooLowResponse())); + assert.equal(Buffer.from((await result)[0]).toString('utf8'), 'too_low'); + assert.equal(refreshes, 0); +}); + +test('failed ticket retry clears the temporary headroom', async () => { + const connection = makeConnection(); + connection._refreshTicketUsage = async () => {}; + connection.createTicketCommand = async () => ['ticketv2']; + connection.sendCommand = async () => { throw new Error('retry failed'); }; + const result = new Promise((resolve, reject) => { + connection.pendingRequests.set(1, { + resolve, + reject, + commandArray: ['ticketv2'], + ticketRetryCount: 0, + }); + }); + + connection._handleData(encodeResponse(1, tooLowResponse())); + await assert.rejects(result, /retry failed/); + assert.equal(connection._ticketHeadroomBytes, 1024); }); test('session responses do not swallow later unsolicited messages on the same session', async () => { @@ -158,7 +223,7 @@ test('session responses do not swallow later unsolicited messages on the same se assert.equal(Buffer.from(message[1][0]).toString('utf8'), 'portsend'); }); -test('too_low repair does not reduce local byte counters', () => { +test('too_low repair discards an unverified local byte target', () => { const connection = makeConnection(); connection.totalConnections = 40; connection.totalBytes = 1304576; @@ -166,10 +231,10 @@ test('too_low repair does not reduce local byte counters', () => { connection.fixResponse(tooLowResponse()); assert.equal(connection.totalConnections, 40); - assert.equal(connection.totalBytes, 1304576); + assert.equal(connection.totalBytes, 135591); }); -test('too_low repair uses node-reported paid byte floor', () => { +test('too_low repair uses only the node-reported paid byte floor', () => { const connection = makeConnection(); connection.totalConnections = 4; connection.totalBytes = 128000; @@ -183,8 +248,8 @@ test('too_low repair uses node-reported paid byte floor', () => { assert.equal(connection.totalConnections, 7); assert.equal(connection.lastAcceptedTicketBytes, 135591); - assert.equal(connection.totalBytes, 140711); - assert.equal(connection.accumulatedBytes, 5120); + assert.equal(connection.totalBytes, 135591); + assert.equal(connection.accumulatedBytes, 0); }); test('ticket update keeps accumulated bytes when ticket is rejected', async () => { @@ -201,3 +266,368 @@ test('ticket update keeps accumulated bytes when ticket is rejected', async () = assert.equal(connection.accumulatedBytes, 1000); assert.equal(connection.lastTicketUpdate, 123); }); + +test('large byte totals are paid in bounded tickets without losing the unpaid target', async () => { + const connection = makeConnection(); + connection.totalBytes = 190_000_000; + connection._waitForServerEthereumAddress = async () => Buffer.alloc(20, 1); + connection.RPC.getEpoch = async () => 687; + const signedBytes = []; + connection.createTicketSignature = async (_server, _connections, bytes) => { + signedBytes.push(bytes); + return Buffer.alloc(65, 1); + }; + + const first = await connection.createTicketCommand(); + assert.equal(first[5], 64_000_000); + connection._recordTicketResponse(first, ['thanks!']); + assert.equal(connection.totalBytes, 190_000_000); + assert.equal(connection.accumulatedBytes, 126_000_000); + + const second = await connection.createTicketCommand(); + assert.equal(second[5], 128_000_000); + connection._recordTicketResponse(second, ['thanks!']); + + const third = await connection.createTicketCommand(); + assert.equal(third[5], 190_000_000); + assert.deepEqual(signedBytes, [first[5], second[5], third[5]]); +}); + +test('too_low repair reconciles against authoritative relay usage without adding local pending bytes', async () => { + const connection = makeConnection(); + connection.totalBytes = 190_000_000; + connection.accumulatedBytes = 120_000_000; + connection._waitForServerEthereumAddress = async () => Buffer.alloc(20, 1); + connection.RPC.getEpoch = async () => 687; + connection.createTicketSignature = async () => Buffer.alloc(65, 1); + + connection.fixResponse(tooLowResponse({ totalConnections: 9, totalBytes: 700_000_000 })); + + assert.equal(connection.lastAcceptedTicketBytes, 700_000_000); + assert.equal(connection.totalBytes, 700_000_000); + connection._supportsRelayUsage = true; + connection._relayUsageGeneration = connection._socketGeneration; + connection._relayUsage = 820_000_000; + const ticket = await connection.createTicketCommand(); + assert.equal(ticket[5], 820_001_024); + assert.equal(connection.totalBytes, 820_001_024); +}); + +test('ticket updates drain a large backlog without exceeding the relay jump limit', async () => { + const connection = makeConnection(); + connection.socket = { destroyed: false, writable: true }; + connection._ready = true; + connection._transportReady = true; + connection.totalBytes = 190_000_000; + connection.accumulatedBytes = 190_000_000; + connection._waitForServerEthereumAddress = async () => Buffer.alloc(20, 1); + connection.RPC.getEpoch = async () => 687; + connection.createTicketSignature = async () => Buffer.alloc(65, 1); + connection._syncMeasuredBytesWithRelay = async () => 0; + connection._refreshTicketUsage = async () => {}; + connection._startTicketUpdateTimer = () => {}; + + let relayFloor = 0; + const acceptedTotals = []; + connection.sendCommand = async (command) => { + const ticketBytes = command[5]; + assert.ok(ticketBytes - relayFloor <= 100_000_000); + assert.ok(ticketBytes > relayFloor); + relayFloor = ticketBytes; + acceptedTotals.push(ticketBytes); + connection._recordTicketResponse(command, ['thanks!']); + return ['thanks!']; + }; + + await connection._updateTicketIfNeeded(true); + for (let index = 0; index < 10 && relayFloor < connection.totalBytes; index += 1) { + await new Promise(setImmediate); + } + + assert.deepEqual(acceptedTotals, [64_000_000, 128_000_000, 190_000_000]); + assert.equal(connection.accumulatedBytes, 0); +}); + +test('epoch rollover does not reuse the previous epoch ticket floor', async () => { + const connection = makeConnection(); + connection.lastAcceptedTicketEpoch = 687; + connection.lastAcceptedTicketBytes = 700_000_000; + connection.totalBytes = 900_000_000; + connection._waitForServerEthereumAddress = async () => Buffer.alloc(20, 1); + connection.RPC.getEpoch = async () => 688; + connection.createTicketSignature = async () => Buffer.alloc(65, 1); + + const first = await connection.createTicketCommand(); + assert.equal(first[2], 688); + assert.equal(first[5], 64_000_000); + assert.equal(connection.lastAcceptedTicketEpoch, 688); + assert.equal(connection.lastAcceptedTicketBytes, 0); + + connection.fixResponse(tooLowResponse({ epoch: 688, totalBytes: 20_000_000 })); + const retry = await connection.createTicketCommand(); + assert.equal(retry[5], 20_000_000); + + // Late responses from an older epoch cannot restore its higher floor. + connection._recordTicketResponse(['ticketv2', 1284, 687, null, 1, 700_000_000], ['thanks!']); + connection.fixResponse(tooLowResponse({ epoch: 687, totalBytes: 700_000_000 })); + assert.equal(connection.lastAcceptedTicketEpoch, 688); + assert.equal(connection.lastAcceptedTicketBytes, 20_000_000); +}); + +for (const beforeResponse of [false, true]) { + test(`hello waits for absolute usage ${beforeResponse ? 'before' : 'after'} its response`, async () => { + const connection = makeConnection(); + stubTicketSigning(connection); + const commands = stubHelloUsage(connection, 10_000_000, { beforeResponse }); + connection.totalBytes = 110_000_000; + connection.accumulatedBytes = 109_000_000; + + assert.equal(await connection._refreshTicketUsage(), 10_000_000); + assert.deepEqual(commands, ['hello']); + assert.equal(connection.totalBytes, 10_001_024); + assert.equal((await connection.createTicketCommand())[5], 10_001_024); + }); +} + +test('hello selects the fresh usage report after an older queued request', async () => { + const connection = makeConnection(); + stubTicketSigning(connection); + connection.sendCommand = async (command) => { + assert.deepEqual(command, ['hello', 1001]); + reportRelayUsage(connection, 700_000_000); // previous epoch, still queued + reportRelayUsage(connection, 20_000_000); // response to this hello + return ['ok']; + }; + + assert.equal(await connection._refreshTicketUsage(), 20_000_000); + assert.equal((await connection.createTicketCommand())[5], 20_001_024); +}); + +test('two idle reconnects keep the same signed byte total and clear stale local bytes', async () => { + let paidBytes = 0; + for (let reconnect = 0; reconnect < 2; reconnect += 1) { + const connection = makeConnection(); + stubTicketSigning(connection); + stubHelloUsage(connection, 10_000_000); + connection.totalBytes = 500_000_000; + connection.accumulatedBytes = 500_000_000; + await connection._refreshTicketUsage(); + const ticket = await connection.createTicketCommand(); + if (reconnect === 0) paidBytes = ticket[5]; + assert.equal(ticket[5], paidBytes); + connection._recordTicketResponse(ticket, ['thanks!']); + assert.equal(connection.accumulatedBytes, 0); + await connection._refreshTicketUsage(); + assert.equal((await connection.createTicketCommand())[5], paidBytes); + assert.equal(connection.accumulatedBytes, 0); + } + assert.equal(paidBytes, 10_001_024); +}); + +test('too_low from a parallel connection does not add local pending bytes to its paid floor', async () => { + const connection = makeConnection(); + stubTicketSigning(connection); + connection.totalBytes = 40_000_000; + connection.accumulatedBytes = 100_000; + connection.fixResponse(tooLowResponse({ totalBytes: 10_101_024 })); + stubHelloUsage(connection, 10_101_024); + + await connection._refreshTicketUsage(); + const ticket = await connection.createTicketCommand(); + assert.equal(ticket[5], 10_101_024); + assert.equal(connection.accumulatedBytes, 0); +}); + +test('too_low with over 64 MB unpaid uses absolute usage rather than doubling the ticket', async () => { + const connection = makeConnection(); + stubTicketSigning(connection); + connection.totalBytes = 220_000_000; + connection.accumulatedBytes = 110_000_000; + connection.fixResponse(tooLowResponse({ totalBytes: 10_000_000 })); + stubHelloUsage(connection, 120_000_000); + + await connection._refreshTicketUsage(); + const ticket = await connection.createTicketCommand(); + assert.equal(ticket[5], 120_001_024); + assert.equal(ticket[5] - 120_000_000, 1024); +}); + +test('active stream too_low retries use fresh usage and bounded temporary margin', async () => { + const connection = makeConnection(); + stubTicketSigning(connection); + const reports = [120_100_000, 120_300_000, 120_700_000]; + let refreshes = 0; + let submissions = 0; + connection._refreshTicketUsage = async () => { + connection._supportsRelayUsage = true; + connection._relayUsageGeneration = connection._socketGeneration; + connection._relayUsage = reports[refreshes]; + refreshes += 1; + }; + connection.sendCommand = (command, options) => { + assert.equal(command[0], 'ticketv2'); + const retryNumber = options.ticketRetryCount; + const currentUsage = [120_300_000, 120_700_000, 121_000_000][submissions]; + assert.equal(retryNumber, submissions + 1); + assert.ok(command[5] - reports[submissions] <= 1024 * 1024); + submissions += 1; + return new Promise((resolve, reject) => { + const requestId = submissions + 1; + connection.pendingRequests.set(requestId, { + resolve, + reject, + commandArray: command, + ticketRetryCount: retryNumber, + }); + const response = submissions < 3 + ? tooLowResponse({ totalBytes: 10_000_000 }) + : ['response', 'thanks!']; + if (submissions < 3) assert.ok(command[5] < currentUsage); + else assert.ok(command[5] >= currentUsage); + setImmediate(() => connection._handleData(encodeResponse(requestId, response))); + }); + }; + + const result = new Promise((resolve, reject) => { + connection.pendingRequests.set(1, { + resolve, + reject, + commandArray: ['ticketv2'], + ticketRetryCount: 0, + }); + }); + connection._handleData(encodeResponse(1, tooLowResponse({ totalBytes: 10_000_000 }))); + assert.equal(Buffer.from((await result)[0]).toString('utf8'), 'thanks!'); + assert.equal(refreshes, 3); + assert.equal(submissions, 3); + assert.equal(connection.lastAcceptedTicketBytes, 121_748_576); + assert.equal(connection._ticketHeadroomBytes, 1024, 'successful ticket resets retry margin'); + connection._relayUsage = 121_000_000; + assert.equal((await connection.createTicketCommand())[5], 121_748_576); +}); + +test('new epoch usage replaces the previous epoch paid floor', async () => { + const connection = makeConnection(); + stubTicketSigning(connection, 688); + connection.lastAcceptedTicketEpoch = 687; + connection.lastAcceptedTicketBytes = 700_000_000; + connection.totalBytes = 900_000_000; + stubHelloUsage(connection, 20_000_000); + + await connection._refreshTicketUsage(); + const ticket = await connection.createTicketCommand(); + assert.equal(ticket[2], 688); + assert.equal(ticket[5], 20_001_024); + assert.equal(connection.lastAcceptedTicketBytes, 0); +}); + +test('unsupported hello wire error probes the paid floor without ambiguous pre-ticket bytes', async () => { + const connection = makeConnection(); + stubTicketSigning(connection); + connection.totalBytes = 110_000_000; + connection.accumulatedBytes = 110_000_000; + const commands = []; + connection.sendCommand = async (command) => { + commands.push(command[0]); + const response = new Promise((resolve, reject) => { + connection.pendingRequests.set(1, { resolve, reject, commandArray: command }); + }); + connection._handleData(encodeResponse(1, ['error', Buffer.from('version not supported')])); + return response; + }; + + await connection._refreshTicketUsage(); + assert.equal((await connection.createTicketCommand())[5], 128000); + connection.fixResponse(tooLowResponse({ totalBytes: 10_000_000 })); + await connection._refreshTicketUsage(); + assert.equal((await connection.createTicketCommand())[5], 10_000_000); + assert.deepEqual(commands, ['hello']); +}); + +test('legacy bytes measurement is used only after an accepted ticket selects the fleet', async () => { + const connection = makeConnection(); + stubTicketSigning(connection); + connection.socket = { destroyed: false }; + connection.lastAcceptedTicketEpoch = 687; + connection.lastAcceptedTicketBytes = 10_000_000; + connection._supportsRelayUsage = false; + connection._ticketAcceptedOnTransport = true; + const commands = []; + let signedUnpaidBytes = 200; // zigzag encoding for +100 + connection.sendCommand = async (command) => { + commands.push(command[0]); + assert.deepEqual(command, ['bytes']); + return [signedUnpaidBytes]; + }; + + await connection._refreshTicketUsage(); + assert.equal((await connection.createTicketCommand())[5], 10_000_100); + signedUnpaidBytes = 3; // zigzag encoding for -1 + await connection._refreshTicketUsage(); + assert.equal((await connection.createTicketCommand())[5], 10_000_000); + assert.equal(connection.accumulatedBytes, 0); + assert.deepEqual(commands, ['bytes', 'bytes']); +}); + +test('hello response without absolute usage fails closed', async () => { + const connection = makeConnection(); + stubTicketSigning(connection); + connection.relayUsageWaitMs = 10; + connection.sendCommand = async (command) => { + assert.deepEqual(command, ['hello', 1001]); + return ['ok']; + }; + + await assert.rejects(connection._refreshTicketUsage(), (error) => error.code === 'DIODE_USAGE_TIMEOUT'); + assert.equal(connection._supportsRelayUsage, null); + assert.equal(connection.listenerCount('relay_usage'), 0); + assert.equal(connection.listenerCount('disconnect'), 0); +}); + +test('disconnect cancels the absolute usage waiter and removes its listeners', async () => { + const connection = makeConnection(); + stubTicketSigning(connection); + connection.sendCommand = async () => ['ok']; + const refresh = connection._refreshTicketUsage(); + for (let turn = 0; turn < 3 && connection.listenerCount('relay_usage') === 0; turn += 1) { + await new Promise(setImmediate); + } + assert.equal(connection.listenerCount('relay_usage'), 1); + connection.emit('disconnect', { generation: connection._socketGeneration }); + await assert.rejects(refresh, (error) => error.code === 'DIODE_DISCONNECTED'); + assert.equal(connection.listenerCount('relay_usage'), 0); + assert.equal(connection.listenerCount('disconnect'), 0); +}); + +test('hello command timeout fails closed without a legacy fallback', async () => { + const connection = makeConnection(); + stubTicketSigning(connection); + connection.sendCommand = async () => { + const error = new Error('hello timed out'); + error.code = 'DIODE_COMMAND_TIMEOUT'; + throw error; + }; + + await assert.rejects(connection._refreshTicketUsage(), (error) => error.code === 'DIODE_COMMAND_TIMEOUT'); + assert.equal(connection._supportsRelayUsage, null); +}); + +test('failed write bytes are removed by the next absolute relay usage report', async () => { + const connection = makeConnection(); + stubTicketSigning(connection); + connection.ticketUpdateThreshold = Number.MAX_SAFE_INTEGER; + connection._ensureConnected = async () => {}; + connection.socket = { + destroyed: false, + writable: true, + write(_message, callback) { callback(new Error('send failed')); }, + }; + + await assert.rejects(connection.sendCommand(['ping', Buffer.alloc(2048)]), /send failed/); + assert.ok(connection.totalBytes > 128000); + assert.equal(connection.pendingRequests.size, 0); + stubHelloUsage(connection, 0); + await connection._refreshTicketUsage(); + assert.equal((await connection.createTicketCommand())[5], 1024); + assert.equal(connection.totalBytes, 1024); +});