Avoid infinite loop on failed claim; allow concurrent thunks

This commit is contained in:
Jon Staab
2024-10-17 12:49:42 -07:00
parent d20d38a838
commit c121997a5f
4 changed files with 58 additions and 59 deletions
+47 -44
View File
@@ -117,7 +117,7 @@ export const publishThunk = (request: ThunkRequest) => {
export const thunkWorker = new Worker<Thunk>()
thunkWorker.addGlobalHandler(async (thunk: Thunk) => {
thunkWorker.addGlobalHandler((thunk: Thunk) => {
let event = thunk.event
// Handle abort immediately if possible
@@ -128,57 +128,60 @@ thunkWorker.addGlobalHandler(async (thunk: Thunk) => {
event = event.wrap
}
// If the event was already signed, leave it alone. Otherwise, sign it now. This is to
// decrease apparent latency in the UI that results from waiting for remote signers
if (!isSignedEvent(event)) {
const signer = getSigner(getSession(event.pubkey))
// Avoid making this function async so multiple publishes can run concurrently
Promise.resolve().then(async () => {
// If the event was already signed, leave it alone. Otherwise, sign it now. This is to
// decrease apparent latency in the UI that results from waiting for remote signers
if (!isSignedEvent(event)) {
const signer = getSigner(getSession(event.pubkey))
if (!signer) {
return console.warn(`No signer found for ${event.pubkey}`)
if (!signer) {
return console.warn(`No signer found for ${event.pubkey}`)
}
event = await signer.sign(event)
}
event = await signer.sign(event)
}
// We're guaranteed to have a signed event at this point
const signedEvent = event as SignedEvent
// We're guaranteed to have a signed event at this point
const signedEvent = event as SignedEvent
// Wait if the thunk is to be delayed
if (thunk.request.delay) {
await sleep(thunk.request.delay)
}
// Skip publishing if aborted
if (thunk.controller.signal.aborted) {
return
}
// Send it off
const pub = publish({event: signedEvent, relays: thunk.request.relays})
// Copy the signature over since we had deferred it
const savedEvent = repository.getEvent(signedEvent.id) as SignedEvent
// The event may already be replaced or deleted
if (savedEvent) {
savedEvent.sig = signedEvent.sig
}
const completed = new Set()
pub.emitter.on("*", async (status: PublishStatus, url: string, message = "") => {
thunk.status.update(assoc(url, {status, message}))
if (status !== PublishStatus.Pending) {
completed.add(url)
// Wait if the thunk is to be delayed
if (thunk.request.delay) {
await sleep(thunk.request.delay)
}
if (status === PublishStatus.Success) {
tracker.track(signedEvent.id, url)
// Skip publishing if aborted
if (thunk.controller.signal.aborted) {
return
}
if (completed.size === thunk.request.relays.length) {
thunk.result.resolve(get(thunk.status))
// Send it off
const pub = publish({event: signedEvent, relays: thunk.request.relays})
// Copy the signature over since we had deferred it
const savedEvent = repository.getEvent(signedEvent.id) as SignedEvent
// The event may already be replaced or deleted
if (savedEvent) {
savedEvent.sig = signedEvent.sig
}
const completed = new Set()
pub.emitter.on("*", async (status: PublishStatus, url: string, message = "") => {
thunk.status.update(assoc(url, {status, message}))
if (status !== PublishStatus.Pending) {
completed.add(url)
}
if (status === PublishStatus.Success) {
tracker.track(signedEvent.id, url)
}
if (completed.size === thunk.request.relays.length) {
thunk.result.resolve(get(thunk.status))
}
})
})
})
+2 -1
View File
@@ -1,4 +1,5 @@
import {Emitter, Worker, sleep} from '@welshman/lib'
import {AUTH_JOIN} from '@welshman/util'
import {ConnectionMeta} from './ConnectionMeta'
import {ConnectionAuth, AuthStatus} from './ConnectionAuth'
import {Socket, isMessage, asMessage} from './Socket'
@@ -43,7 +44,7 @@ export class Connection extends Emitter {
}
// Allow relay requests through
if (verb === 'EVENT' && extra[0].kind === 28934) {
if (verb === 'EVENT' && extra[0].kind === AUTH_JOIN) {
return false
}
-2
View File
@@ -46,8 +46,6 @@ export class ConnectionAuth {
const [id, ok, message] = extra
if (id === this.request) {
this.challenge = undefined
this.request = undefined
this.message = message
this.status = ok ? Ok : Forbidden
}
+9 -12
View File
@@ -1,3 +1,4 @@
import {AUTH_JOIN} from '@welshman/util'
import type {SignedEvent, Filter} from '@welshman/util'
import type {Message} from './Socket'
import type {Connection} from './Connection'
@@ -86,20 +87,16 @@ export class ConnectionMeta {
}
onReceiveOk([verb, eventId, ok, notice]: Message) {
const pub = this.pendingPublishes.get(eventId)
if (!pub) return
// Re-enqueue pending events when auth challenge is received
if (notice?.startsWith('auth-required:')) {
const pub = this.pendingPublishes.get(eventId)
if (pub) {
this.cxn.send(['EVENT', pub.event])
}
}
const publish = this.pendingPublishes.get(eventId)
if (publish) {
if (notice?.startsWith('auth-required:') && pub.event.kind !== AUTH_JOIN) {
this.cxn.send(['EVENT', pub.event])
} else {
this.responseCount++
this.responseTimer += Date.now() - publish.sent
this.responseTimer += Date.now() - pub.sent
this.pendingPublishes.delete(eventId)
}
}