enableReconnect() always update the filters to the time of the last received event on each subscription.

This commit is contained in:
fiatjaf
2025-12-05 13:21:20 -03:00
parent de7d459f6f
commit 85c964be3d
6 changed files with 16 additions and 76 deletions

View File

@@ -36,7 +36,7 @@ export class AbstractSimplePool {
public verifyEvent: Nostr['verifyEvent'] public verifyEvent: Nostr['verifyEvent']
public enablePing: boolean | undefined public enablePing: boolean | undefined
public enableReconnect: boolean | ((filters: Filter[]) => Filter[]) | undefined public enableReconnect: boolean
public automaticallyAuth?: (relayURL: string) => null | ((event: EventTemplate) => Promise<VerifiedEvent>) public automaticallyAuth?: (relayURL: string) => null | ((event: EventTemplate) => Promise<VerifiedEvent>)
public trustedRelayURLs: Set<string> = new Set() public trustedRelayURLs: Set<string> = new Set()
@@ -46,7 +46,7 @@ export class AbstractSimplePool {
this.verifyEvent = opts.verifyEvent this.verifyEvent = opts.verifyEvent
this._WebSocket = opts.websocketImplementation this._WebSocket = opts.websocketImplementation
this.enablePing = opts.enablePing this.enablePing = opts.enablePing
this.enableReconnect = opts.enableReconnect this.enableReconnect = opts.enableReconnect || false
this.automaticallyAuth = opts.automaticallyAuth this.automaticallyAuth = opts.automaticallyAuth
} }

View File

@@ -16,7 +16,7 @@ export type AbstractRelayConstructorOptions = {
verifyEvent: Nostr['verifyEvent'] verifyEvent: Nostr['verifyEvent']
websocketImplementation?: typeof WebSocket websocketImplementation?: typeof WebSocket
enablePing?: boolean enablePing?: boolean
enableReconnect?: boolean | ((filters: Filter[]) => Filter[]) enableReconnect?: boolean
} }
export class SendingOnClosedConnection extends Error { export class SendingOnClosedConnection extends Error {
@@ -42,7 +42,7 @@ export class AbstractRelay {
public resubscribeBackoff: number[] = [10000, 10000, 10000, 20000, 20000, 30000, 60000] public resubscribeBackoff: number[] = [10000, 10000, 10000, 20000, 20000, 30000, 60000]
public openSubs: Map<string, Subscription> = new Map() public openSubs: Map<string, Subscription> = new Map()
public enablePing: boolean | undefined public enablePing: boolean | undefined
public enableReconnect: boolean | ((filters: Filter[]) => Filter[]) public enableReconnect: boolean
private connectionTimeoutHandle: ReturnType<typeof setTimeout> | undefined private connectionTimeoutHandle: ReturnType<typeof setTimeout> | undefined
private reconnectTimeoutHandle: ReturnType<typeof setTimeout> | undefined private reconnectTimeoutHandle: ReturnType<typeof setTimeout> | undefined
private pingTimeoutHandle: ReturnType<typeof setTimeout> | undefined private pingTimeoutHandle: ReturnType<typeof setTimeout> | undefined
@@ -166,8 +166,12 @@ export class AbstractRelay {
// resubscribe to all open subscriptions // resubscribe to all open subscriptions
for (const sub of this.openSubs.values()) { for (const sub of this.openSubs.values()) {
sub.eosed = false sub.eosed = false
if (isReconnection && typeof this.enableReconnect === 'function') { if (isReconnection) {
sub.filters = this.enableReconnect(sub.filters) for (let f = 0; f < sub.filters.length; f++) {
if (sub.lastEmitted) {
sub.filters[f].since = sub.lastEmitted + 1
}
}
} }
sub.fire() sub.fire()
} }
@@ -299,6 +303,7 @@ export class AbstractRelay {
if (this.verifyEvent(event) && matchFilters(so.filters, event)) { if (this.verifyEvent(event) && matchFilters(so.filters, event)) {
so.onevent(event) so.onevent(event)
} }
if (!so.lastEmitted || so.lastEmitted < event.created_at) so.lastEmitted = event.created_at
return return
} }
case 'COUNT': { case 'COUNT': {
@@ -469,6 +474,7 @@ export class Subscription {
public readonly relay: AbstractRelay public readonly relay: AbstractRelay
public readonly id: string public readonly id: string
public lastEmitted: number | undefined
public closed: boolean = false public closed: boolean = false
public eosed: boolean = false public eosed: boolean = false
public filters: Filter[] public filters: Filter[]

View File

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

View File

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

View File

@@ -306,12 +306,9 @@ test('reconnect on disconnect in pool', async () => {
test('reconnect with filter update in pool', async () => { test('reconnect with filter update in pool', async () => {
const mockRelay = mockRelays[0] const mockRelay = mockRelays[0]
const newSince = Math.floor(Date.now() / 1000)
pool = new SimplePool({ pool = new SimplePool({
enablePing: true, enablePing: true,
enableReconnect: filters => { enableReconnect: true,
return filters.map(f => ({ ...f, since: newSince }))
},
}) })
const relay = await pool.ensureRelay(mockRelay.url) const relay = await pool.ensureRelay(mockRelay.url)
relay.pingTimeout = 50 relay.pingTimeout = 50
@@ -364,7 +361,7 @@ test('reconnect with filter update in pool', async () => {
expect(closes).toBe(1) expect(closes).toBe(1)
// check if filter was updated // check if filter was updated
expect(sub.filters[0].since).toBe(newSince) expect(sub.filters[0].since).toBeGreaterThan(1)
}) })
test('track relays when publishing', async () => { test('track relays when publishing', async () => {

View File

@@ -336,66 +336,3 @@ test('reconnect on disconnect', async () => {
expect(relay.connected).toBeTrue() expect(relay.connected).toBeTrue()
expect(closes).toBe(1) // should not have closed again expect(closes).toBe(1) // should not have closed again
}) })
test('reconnect with filter update', async () => {
const mockRelay = new MockRelay()
const newSince = Math.floor(Date.now() / 1000)
const relay = new Relay(mockRelay.url, {
enablePing: true,
enableReconnect: filters => {
return filters.map(f => ({ ...f, since: newSince }))
},
})
relay.pingTimeout = 50
relay.pingFrequency = 50
relay.resubscribeBackoff = [50, 100]
let closes = 0
relay.onclose = () => {
closes++
}
await relay.connect()
expect(relay.connected).toBeTrue()
const sub = relay.subscribe([{ kinds: [1], since: 0 }], { onevent: () => {} })
expect(sub.filters[0].since).toBe(0)
// wait for the first ping to succeed
await new Promise(resolve => setTimeout(resolve, 75))
expect(closes).toBe(0)
// now make it unresponsive
mockRelay.unresponsive = true
// wait for the second ping to fail, which will trigger a close
await new Promise(resolve => {
const interval = setInterval(() => {
if (closes > 0) {
clearInterval(interval)
resolve(null)
}
}, 10)
})
expect(closes).toBe(1)
expect(relay.connected).toBeFalse()
// now make it responsive again
mockRelay.unresponsive = false
// wait for reconnect
await new Promise(resolve => {
const interval = setInterval(() => {
if (relay.connected) {
clearInterval(interval)
resolve(null)
}
}, 10)
})
expect(relay.connected).toBeTrue()
expect(closes).toBe(1)
// check if filter was updated
expect(sub.filters[0].since).toBe(newSince)
})