Compare commits

...

2 Commits

Author SHA1 Message Date
fiatjaf
ccb9641fb9 pool: maxWaitForConnection parameter.
this was so obvious.
2026-01-31 00:27:55 -03:00
fiatjaf
b624ad4059 pool: hooks to notify when a relay fails to connect, then ask whether a connection should be attempted. 2026-01-30 17:35:46 -03:00
5 changed files with 65 additions and 9 deletions

View File

@@ -11,6 +11,7 @@ import { normalizeURL } from './utils.ts'
import type { Event, EventTemplate, Nostr, VerifiedEvent } from './core.ts'
import { type Filter } from './filter.ts'
import { alwaysTrue } from './helpers.ts'
import { Relay } from './relay.ts'
export type SubCloser = { close: (reason?: string) => void }
@@ -19,6 +20,14 @@ export type AbstractPoolConstructorOptions = AbstractRelayConstructorOptions & {
// in case that relay shouldn't be authenticated against
// or a function to sign the AUTH event template otherwise (that function may still throw in case of failure)
automaticallyAuth?: (relayURL: string) => null | ((event: EventTemplate) => Promise<VerifiedEvent>)
// onRelayConnectionFailure is called with the URL of a relay that failed the initial connection
onRelayConnectionFailure?: (url: string) => void
// allowConnectingToRelay takes a relay URL and the operation being performed
// return false to skip connecting to that relay
allowConnectingToRelay?: (url: string, operation: ['read', Filter[]] | ['write', Event]) => boolean
// maxWaitForConnection takes a number in milliseconds that will be given to ensureRelay such that we
// don't get stuck forever when attempting to connect to a relay, it is 3000 (3 seconds) by default
maxWaitForConnection: number
}
export type SubscribeManyParams = Omit<SubscriptionParams, 'onclose'> & {
@@ -40,6 +49,9 @@ export class AbstractSimplePool {
public enableReconnect: boolean
public automaticallyAuth?: (relayURL: string) => null | ((event: EventTemplate) => Promise<VerifiedEvent>)
public trustedRelayURLs: Set<string> = new Set()
public onRelayConnectionFailure?: (url: string) => void
public allowConnectingToRelay?: (url: string, operation: ['read', Filter[]] | ['write', Event]) => boolean
public maxWaitForConnection: number
private _WebSocket?: typeof WebSocket
@@ -49,6 +61,9 @@ export class AbstractSimplePool {
this.enablePing = opts.enablePing
this.enableReconnect = opts.enableReconnect || false
this.automaticallyAuth = opts.automaticallyAuth
this.onRelayConnectionFailure = opts.onRelayConnectionFailure
this.allowConnectingToRelay = opts.allowConnectingToRelay
this.maxWaitForConnection = opts.maxWaitForConnection || 3000
}
async ensureRelay(
@@ -181,13 +196,22 @@ export class AbstractSimplePool {
// open a subscription in all given relays
const allOpened = Promise.all(
groupedRequests.map(async ({ url, filters }, i) => {
if (this.allowConnectingToRelay?.(url, ['read', filters]) === false) {
handleClose(i, 'connection skipped by allowConnectingToRelay')
return
}
let relay: AbstractRelay
try {
relay = await this.ensureRelay(url, {
connectionTimeout: params.maxWait ? Math.max(params.maxWait * 0.8, params.maxWait - 1000) : undefined,
connectionTimeout:
this.maxWaitForConnection < (params.maxWait || 0)
? Math.max(params.maxWait! * 0.8, params.maxWait! - 1000)
: this.maxWaitForConnection,
abort: params.abort,
})
} catch (err) {
this.onRelayConnectionFailure?.(url)
handleClose(i, (err as any)?.message || String(err))
return
}
@@ -298,7 +322,11 @@ export class AbstractSimplePool {
publish(
relays: string[],
event: Event,
options?: { onauth?: (evt: EventTemplate) => Promise<VerifiedEvent> },
params?: {
onauth?: (evt: EventTemplate) => Promise<VerifiedEvent>
maxWait?: number
abort?: AbortSignal
},
): Promise<string>[] {
return relays.map(normalizeURL).map(async (url, i, arr) => {
if (arr.indexOf(url) !== i) {
@@ -306,12 +334,29 @@ export class AbstractSimplePool {
return Promise.reject('duplicate url')
}
let r = await this.ensureRelay(url)
if (this.allowConnectingToRelay?.(url, ['write', event]) === false) {
return Promise.reject('connection skipped by allowConnectingToRelay')
}
let r: Relay
try {
r = await this.ensureRelay(url, {
connectionTimeout:
this.maxWaitForConnection < (params?.maxWait || 0)
? Math.max(params!.maxWait! * 0.8, params!.maxWait! - 1000)
: this.maxWaitForConnection,
abort: params?.abort,
})
} catch (err) {
this.onRelayConnectionFailure?.(url)
return String('connection failure: ' + String(err))
}
return r
.publish(event)
.catch(async err => {
if (err instanceof Error && err.message.startsWith('auth-required: ') && options?.onauth) {
await r.auth(options.onauth)
if (err instanceof Error && err.message.startsWith('auth-required: ') && params?.onauth) {
await r.auth(params.onauth)
return r.publish(event) // retry
}
throw err

View File

@@ -145,7 +145,7 @@ export class AbstractRelay {
reject('connection timed out')
this.connectionPromise = undefined
this.onclose?.()
this.closeAllSubscriptions('relay connection timed out')
this.handleHardClose('relay connection timed out')
}, opts.timeout)
}
@@ -153,8 +153,17 @@ export class AbstractRelay {
opts.abort.onabort = reject
}
const connectionFailed = () => {
clearTimeout(connectionTimeoutHandle)
reject('connection failed')
this.connectionPromise = undefined
this.onclose?.()
this.handleHardClose('relay connection failed')
}
try {
this.ws = new this._WebSocket(this.url)
this.ws.addEventListener('error', connectionFailed)
} catch (err) {
clearTimeout(connectionTimeoutHandle)
reject(err)
@@ -162,6 +171,8 @@ export class AbstractRelay {
}
this.ws.onopen = () => {
this.ws?.removeEventListener('error', connectionFailed)
if (this.reconnectTimeoutHandle) {
clearTimeout(this.reconnectTimeoutHandle)
this.reconnectTimeoutHandle = undefined

View File

@@ -1,6 +1,6 @@
{
"name": "@nostr/tools",
"version": "2.22.0",
"version": "2.22.2",
"exports": {
".": "./index.ts",
"./core": "./core.ts",

View File

@@ -1,7 +1,7 @@
{
"type": "module",
"name": "nostr-tools",
"version": "2.22.0",
"version": "2.22.2",
"description": "Tools for making a Nostr client.",
"repository": {
"type": "git",

View File

@@ -15,7 +15,7 @@ export function useWebSocketImplementation(websocketImplementation: any) {
export class SimplePool extends AbstractSimplePool {
constructor(options?: Pick<AbstractPoolConstructorOptions, 'enablePing' | 'enableReconnect'>) {
super({ verifyEvent, websocketImplementation: _WebSocket, ...options })
super({ verifyEvent, websocketImplementation: _WebSocket, maxWaitForConnection: 3000, ...options })
}
}