Fix duplicate answer processing race condition (#6)

Add polling guard to prevent concurrent pollInternal() execution.
The setInterval callback doesn't await the async pollInternal(),
which could cause multiple polls to process the same answer before
lastPollTimestamp is updated, resulting in "Called in wrong state:
stable" errors from setRemoteDescription().

README.md

@@ -49,8 +49,8 @@ const rondevu = await Rondevu.connect({
 await rondevu.publishService({
   service: 'chat:1.0.0',
   maxOffers: 5,  // Maintain up to 5 concurrent offers
-  offerFactory: async (rtcConfig) => {
-    const pc = new RTCPeerConnection(rtcConfig)
+  offerFactory: async (pc) => {
+    // pc is created by Rondevu with ICE handlers already attached
     const dc = pc.createDataChannel('chat')
 
     dc.addEventListener('open', () => {
@@ -64,7 +64,7 @@ await rondevu.publishService({
 
     const offer = await pc.createOffer()
     await pc.setLocalDescription(offer)
-    return { pc, dc, offer }
+    return { dc, offer }
   }
 })
 

package-lock.json

@@ -13,6 +13,7 @@
       },
       "devDependencies": {
         "@eslint/js": "^9.39.1",
+        "@types/node": "^25.0.2",
         "@typescript-eslint/eslint-plugin": "^8.48.1",
         "@typescript-eslint/parser": "^8.48.1",
         "eslint": "^9.39.1",
@@ -1075,6 +1076,16 @@
       "dev": true,
       "license": "MIT"
     },
+    "node_modules/@types/node": {
+      "version": "25.0.2",
+      "resolved": "https://registry.npmjs.org/@types/node/-/node-25.0.2.tgz",
+      "integrity": "sha512-gWEkeiyYE4vqjON/+Obqcoeffmk0NF15WSBwSs7zwVA2bAbTaE0SJ7P0WNGoJn8uE7fiaV5a7dKYIJriEqOrmA==",
+      "dev": true,
+      "license": "MIT",
+      "dependencies": {
+        "undici-types": "~7.16.0"
+      }
+    },
     "node_modules/@typescript-eslint/eslint-plugin": {
       "version": "8.48.1",
       "resolved": "https://registry.npmjs.org/@typescript-eslint/eslint-plugin/-/eslint-plugin-8.48.1.tgz",
@@ -2828,6 +2839,13 @@
         "node": ">=14.17"
       }
     },
+    "node_modules/undici-types": {
+      "version": "7.16.0",
+      "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-7.16.0.tgz",
+      "integrity": "sha512-Zz+aZWSj8LE6zoxD+xrjh4VfkIG8Ya6LvYkZqtUQGJPZjYl53ypCaUwWqo7eI0x66KBGeRo+mlBEkMSeSZ38Nw==",
+      "dev": true,
+      "license": "MIT"
+    },
     "node_modules/update-browserslist-db": {
       "version": "1.2.2",
       "resolved": "https://registry.npmjs.org/update-browserslist-db/-/update-browserslist-db-1.2.2.tgz",

package.json

@@ -25,6 +25,7 @@
   "license": "MIT",
   "devDependencies": {
     "@eslint/js": "^9.39.1",
+    "@types/node": "^25.0.2",
     "@typescript-eslint/eslint-plugin": "^8.48.1",
     "@typescript-eslint/parser": "^8.48.1",
     "eslint": "^9.39.1",

src/api.ts

@@ -39,6 +39,7 @@ export interface Service {
 
 export interface IceCandidate {
     candidate: RTCIceCandidateInit | null
+    role: 'offerer' | 'answerer'
     createdAt: number
 }
 

src/index.ts

@@ -3,7 +3,7 @@
  * WebRTC peer signaling client
  */
 
-export { Rondevu } from './rondevu.js'
+export { Rondevu, RondevuError, NetworkError, ValidationError, ConnectionError } from './rondevu.js'
 export { RondevuAPI } from './api.js'
 export { RpcBatcher } from './rpc-batcher.js'
 
@@ -32,7 +32,11 @@ export type {
     ConnectToServiceOptions,
     ConnectionContext,
     OfferContext,
-    OfferFactory
+    OfferFactory,
+    ActiveOffer,
+    FindServiceOptions,
+    ServiceResult,
+    PaginatedServiceResult
 } from './rondevu.js'
 
 export type { CryptoAdapter } from './crypto-adapter.js'

src/node-crypto-adapter.ts

@@ -81,13 +81,11 @@ export class NodeCryptoAdapter implements CryptoAdapter {
 
     bytesToBase64(bytes: Uint8Array): string {
         // Node.js Buffer provides native base64 encoding
-        // @ts-expect-error - Buffer is available in Node.js but not in browser TypeScript definitions
         return Buffer.from(bytes).toString('base64')
     }
 
     base64ToBytes(base64: string): Uint8Array {
         // Node.js Buffer provides native base64 decoding
-        // @ts-expect-error - Buffer is available in Node.js but not in browser TypeScript definitions
         return new Uint8Array(Buffer.from(base64, 'base64'))
     }
 

src/rondevu.ts

@@ -1,5 +1,6 @@
 import { RondevuAPI, Keypair, IceCandidate, BatcherOptions } from './api.js'
 import { CryptoAdapter } from './crypto-adapter.js'
+import { EventEmitter } from 'events'
 
 // ICE server preset names
 export type IceServerPreset = 'ipv4-turn' | 'hostname-turns' | 'google-stun' | 'relay-only'
@@ -63,12 +64,19 @@ export interface RondevuOptions {
 }
 
 export interface OfferContext {
-    pc: RTCPeerConnection
     dc?: RTCDataChannel
     offer: RTCSessionDescriptionInit
 }
 
-export type OfferFactory = (rtcConfig: RTCConfiguration) => Promise<OfferContext>
+/**
+ * Factory function for creating WebRTC offers.
+ * Rondevu creates the RTCPeerConnection and passes it to the factory,
+ * allowing ICE candidate handlers to be set up before setLocalDescription() is called.
+ *
+ * @param pc - The RTCPeerConnection created by Rondevu (already configured with ICE servers)
+ * @returns Promise containing the data channel (optional) and offer SDP
+ */
+export type OfferFactory = (pc: RTCPeerConnection) => Promise<OfferContext>
 
 export interface PublishServiceOptions {
     service: string // Service name and version (e.g., "chat:2.0.0") - username will be auto-appended
@@ -93,7 +101,7 @@ export interface ConnectToServiceOptions {
     rtcConfig?: RTCConfiguration  // Optional: override default ICE servers
 }
 
-interface ActiveOffer {
+export interface ActiveOffer {
     offerId: string
     serviceFqn: string
     pc: RTCPeerConnection
@@ -102,6 +110,73 @@ interface ActiveOffer {
     createdAt: number
 }
 
+export interface FindServiceOptions {
+    mode?: 'direct' | 'random' | 'paginated'  // Default: 'direct' if serviceFqn has username, 'random' otherwise
+    limit?: number  // For paginated mode (default: 10)
+    offset?: number  // For paginated mode (default: 0)
+}
+
+export interface ServiceResult {
+    serviceId: string
+    username: string
+    serviceFqn: string
+    offerId: string
+    sdp: string
+    createdAt: number
+    expiresAt: number
+}
+
+export interface PaginatedServiceResult {
+    services: ServiceResult[]
+    count: number
+    limit: number
+    offset: number
+}
+
+/**
+ * Base error class for Rondevu errors
+ */
+export class RondevuError extends Error {
+    constructor(message: string, public context?: Record<string, any>) {
+        super(message)
+        this.name = 'RondevuError'
+        Object.setPrototypeOf(this, RondevuError.prototype)
+    }
+}
+
+/**
+ * Network-related errors (API calls, connectivity)
+ */
+export class NetworkError extends RondevuError {
+    constructor(message: string, context?: Record<string, any>) {
+        super(message, context)
+        this.name = 'NetworkError'
+        Object.setPrototypeOf(this, NetworkError.prototype)
+    }
+}
+
+/**
+ * Validation errors (invalid input, malformed data)
+ */
+export class ValidationError extends RondevuError {
+    constructor(message: string, context?: Record<string, any>) {
+        super(message, context)
+        this.name = 'ValidationError'
+        Object.setPrototypeOf(this, ValidationError.prototype)
+    }
+}
+
+/**
+ * WebRTC connection errors (peer connection failures, ICE issues)
+ */
+export class ConnectionError extends RondevuError {
+    constructor(message: string, context?: Record<string, any>) {
+        super(message, context)
+        this.name = 'ConnectionError'
+        Object.setPrototypeOf(this, ConnectionError.prototype)
+    }
+}
+
 /**
  * Rondevu - Complete WebRTC signaling client
  *
@@ -135,12 +210,12 @@ interface ActiveOffer {
  * await rondevu.publishService({
  *   service: 'chat:2.0.0',
  *   maxOffers: 5,  // Maintain up to 5 concurrent offers
- *   offerFactory: async (rtcConfig) => {
- *     const pc = new RTCPeerConnection(rtcConfig)
+ *   offerFactory: async (pc) => {
+ *     // pc is created by Rondevu with ICE handlers already attached
  *     const dc = pc.createDataChannel('chat')
  *     const offer = await pc.createOffer()
  *     await pc.setLocalDescription(offer)
- *     return { pc, dc, offer }
+ *     return { dc, offer }
  *   }
  * })
  *
@@ -156,7 +231,7 @@ interface ActiveOffer {
  * rondevu.stopFilling()
  * ```
  */
-export class Rondevu {
+export class Rondevu extends EventEmitter {
     // Constants
     private static readonly DEFAULT_TTL_MS = 300000  // 5 minutes
     private static readonly POLLING_INTERVAL_MS = 1000  // 1 second
@@ -184,6 +259,7 @@ export class Rondevu {
     private filling = false
     private pollingInterval: ReturnType<typeof setInterval> | null = null
     private lastPollTimestamp = 0
+    private isPolling = false  // Guard against concurrent poll execution
 
     private constructor(
         apiUrl: string,
@@ -197,6 +273,7 @@ export class Rondevu {
         rtcPeerConnection?: typeof RTCPeerConnection,
         rtcIceCandidate?: typeof RTCIceCandidate
     ) {
+        super()
         this.apiUrl = apiUrl
         this.username = username
         this.keypair = keypair
@@ -337,15 +414,15 @@ export class Rondevu {
 
     /**
      * Default offer factory - creates a simple data channel connection
+     * The RTCPeerConnection is created by Rondevu and passed in
      */
-    private async defaultOfferFactory(rtcConfig: RTCConfiguration): Promise<OfferContext> {
-        const pc = new RTCPeerConnection(rtcConfig)
+    private async defaultOfferFactory(pc: RTCPeerConnection): Promise<OfferContext> {
         const dc = pc.createDataChannel('default')
 
         const offer = await pc.createOffer()
         await pc.setLocalDescription(offer)
 
-        return { pc, dc, offer }
+        return { dc, offer }
     }
 
     /**
@@ -375,6 +452,10 @@ export class Rondevu {
 
     /**
      * Set up ICE candidate handler to send candidates to the server
+     *
+     * Note: This is used by connectToService() where the offerId is already known.
+     * For createOffer(), we use inline ICE handling with early candidate queuing
+     * since the offerId isn't available until after the factory completes.
      */
     private setupIceCandidateHandler(
         pc: RTCPeerConnection,
@@ -391,6 +472,9 @@ export class Rondevu {
                         ? event.candidate.toJSON()
                         : event.candidate
 
+                    // Emit local ICE candidate event
+                    this.emit('ice:candidate:local', offerId, candidateData)
+
                     await this.api.addOfferIceCandidates(
                         serviceFqn,
                         offerId,
@@ -415,23 +499,20 @@ export class Rondevu {
             iceServers: this.iceServers
         }
 
-        this.debug('Creating new offer...')
-
-        // Create the offer using the factory
-        // Note: The factory may call setLocalDescription() which triggers ICE gathering
-        const { pc, dc, offer } = await this.offerFactory(rtcConfig)
-
         // Auto-append username to service
         const serviceFqn = `${this.currentService}@${this.username}`
 
-        // Queue to buffer ICE candidates generated before we have the offerId
-        // This fixes the race condition where ICE candidates are lost because
-        // they're generated before we can set up the handler with the offerId
-        const earlyIceCandidates: RTCIceCandidateInit[] = []
-        let offerId: string | null = null
+        this.debug('Creating new offer...')
+
+        // 1. Create the RTCPeerConnection - Rondevu controls this to set up handlers early
+        const pc = new RTCPeerConnection(rtcConfig)
+
+        // 2. Set up ICE candidate handler with queuing BEFORE the factory runs
+        // This ensures we capture all candidates, even those generated immediately
+        // when setLocalDescription() is called in the factory
+        const earlyIceCandidates: RTCIceCandidateInit[] = []
+        let offerId: string | undefined
 
-        // Set up a queuing ICE candidate handler immediately after getting the pc
-        // This captures any candidates that fire before we have the offerId
         pc.onicecandidate = async (event) => {
             if (event.candidate) {
                 // Handle both browser and Node.js (wrtc) environments
@@ -439,6 +520,11 @@ export class Rondevu {
                     ? event.candidate.toJSON()
                     : event.candidate
 
+                // Emit local ICE candidate event
+                if (offerId) {
+                    this.emit('ice:candidate:local', offerId, candidateData)
+                }
+
                 if (offerId) {
                     // We have the offerId, send directly
                     try {
@@ -454,7 +540,22 @@ export class Rondevu {
             }
         }
 
-        // Publish to server
+        // 3. Call the factory with the pc - factory creates data channel and offer
+        // When factory calls setLocalDescription(), ICE gathering starts and
+        // candidates are captured by the handler we set up above
+        let dc: RTCDataChannel | undefined
+        let offer: RTCSessionDescriptionInit
+        try {
+            const factoryResult = await this.offerFactory(pc)
+            dc = factoryResult.dc
+            offer = factoryResult.offer
+        } catch (err) {
+            // Clean up the connection if factory fails
+            pc.close()
+            throw err
+        }
+
+        // 4. Publish to server to get offerId
         const result = await this.api.publishService({
             serviceFqn,
             offers: [{ sdp: offer.sdp! }],
@@ -465,7 +566,7 @@ export class Rondevu {
 
         offerId = result.offers[0].offerId
 
-        // Store active offer
+        // 5. Store active offer
         this.activeOffers.set(offerId, {
             offerId,
             serviceFqn,
@@ -476,8 +577,17 @@ export class Rondevu {
         })
 
         this.debug(`Offer created: ${offerId}`)
+        this.emit('offer:created', offerId, serviceFqn)
 
-        // Send any queued early ICE candidates
+        // Set up data channel open handler (offerer side)
+        if (dc) {
+            dc.onopen = () => {
+                this.debug(`Data channel opened for offer ${offerId}`)
+                this.emit('connection:opened', offerId, dc)
+            }
+        }
+
+        // 6. Send any queued early ICE candidates
         if (earlyIceCandidates.length > 0) {
             this.debug(`Sending ${earlyIceCandidates.length} early ICE candidates`)
             try {
@@ -487,11 +597,12 @@ export class Rondevu {
             }
         }
 
-        // Monitor connection state
+        // 7. Monitor connection state
         pc.onconnectionstatechange = () => {
             this.debug(`Offer ${offerId} connection state: ${pc.connectionState}`)
 
             if (pc.connectionState === 'failed' || pc.connectionState === 'closed') {
+                this.emit('connection:closed', offerId!)
                 this.activeOffers.delete(offerId!)
                 this.fillOffers()  // Try to replace failed offer
             }
@@ -524,6 +635,13 @@ export class Rondevu {
     private async pollInternal(): Promise<void> {
         if (!this.filling) return
 
+        // Prevent concurrent poll execution to avoid duplicate answer processing
+        if (this.isPolling) {
+            this.debug('Poll already in progress, skipping')
+            return
+        }
+
+        this.isPolling = true
         try {
             const result = await this.api.poll(this.lastPollTimestamp)
 
@@ -540,6 +658,7 @@ export class Rondevu {
 
                     activeOffer.answered = true
                     this.lastPollTimestamp = answer.answeredAt
+                    this.emit('offer:answered', answer.offerId, answer.answererId)
 
                     // Create replacement offer
                     this.fillOffers()
@@ -554,6 +673,7 @@ export class Rondevu {
 
                     for (const item of answererCandidates) {
                         if (item.candidate) {
+                            this.emit('ice:candidate:remote', offerId, item.candidate, item.role)
                             await activeOffer.pc.addIceCandidate(new RTCIceCandidate(item.candidate))
                             this.lastPollTimestamp = Math.max(this.lastPollTimestamp, item.createdAt)
                         }
@@ -562,6 +682,8 @@ export class Rondevu {
             }
         } catch (err) {
             console.error('[Rondevu] Polling error:', err)
+        } finally {
+            this.isPolling = false
         }
     }
 
@@ -598,6 +720,7 @@ export class Rondevu {
     stopFilling(): void {
         this.debug('Stopping offer filling and polling')
         this.filling = false
+        this.isPolling = false  // Reset polling guard
 
         // Stop polling
         if (this.pollingInterval) {
@@ -615,6 +738,51 @@ export class Rondevu {
         this.activeOffers.clear()
     }
 
+    /**
+     * Get the count of active offers
+     * @returns Number of active offers
+     */
+    getOfferCount(): number {
+        return this.activeOffers.size
+    }
+
+    /**
+     * Check if an offer is currently connected
+     * @param offerId - The offer ID to check
+     * @returns True if the offer exists and has been answered
+     */
+    isConnected(offerId: string): boolean {
+        const offer = this.activeOffers.get(offerId)
+        return offer ? offer.answered : false
+    }
+
+    /**
+     * Disconnect all active offers
+     * Similar to stopFilling() but doesn't stop the polling/filling process
+     */
+    async disconnectAll(): Promise<void> {
+        this.debug('Disconnecting all offers')
+        for (const [offerId, offer] of this.activeOffers.entries()) {
+            this.debug(`Closing offer ${offerId}`)
+            offer.dc?.close()
+            offer.pc.close()
+        }
+        this.activeOffers.clear()
+    }
+
+    /**
+     * Get the current service status
+     * @returns Object with service state information
+     */
+    getServiceStatus(): { active: boolean; offerCount: number; maxOffers: number; filling: boolean } {
+        return {
+            active: this.currentService !== null,
+            offerCount: this.activeOffers.size,
+            maxOffers: this.maxOffers,
+            filling: this.filling
+        }
+    }
+
     /**
      * Resolve the full service FQN from various input options
      * Supports direct FQN, service+username, or service discovery
@@ -629,7 +797,7 @@ export class Rondevu {
         } else if (service) {
             // Discovery mode - get random service
             this.debug(`Discovering service: ${service}`)
-            const discovered = await this.discoverService(service)
+            const discovered = await this.findService(service) as ServiceResult
             return discovered.serviceFqn
         } else {
             throw new Error('Either serviceFqn or service must be provided')
@@ -656,6 +824,7 @@ export class Rondevu {
                 )
                 for (const item of result.candidates) {
                     if (item.candidate) {
+                        this.emit('ice:candidate:remote', offerId, item.candidate, item.role)
                         await pc.addIceCandidate(new RTCIceCandidate(item.candidate))
                         lastIceTimestamp = item.createdAt
                     }
@@ -725,6 +894,7 @@ export class Rondevu {
             pc.ondatachannel = (event) => {
                 this.debug('Data channel received from offerer')
                 dc = event.channel
+                this.emit('connection:opened', serviceData.offerId, dc)
                 resolve(dc)
             }
         })
@@ -796,56 +966,45 @@ export class Rondevu {
     // ============================================
 
     /**
-     * Get service by FQN (with username) - Direct lookup
-     * Example: chat:1.0.0@alice
+     * Find a service - unified discovery method
+     *
+     * Replaces getService(), discoverService(), and discoverServices() with a single method.
+     *
+     * @param serviceFqn - Service identifier (e.g., 'chat:1.0.0' or 'chat:1.0.0@alice')
+     * @param options - Discovery options
+     *
+     * @example
+     * ```typescript
+     * // Direct lookup (has username)
+     * const service = await rondevu.findService('chat:1.0.0@alice')
+     *
+     * // Random discovery (no username)
+     * const service = await rondevu.findService('chat:1.0.0')
+     *
+     * // Paginated discovery
+     * const result = await rondevu.findService('chat:1.0.0', {
+     *   mode: 'paginated',
+     *   limit: 20,
+     *   offset: 0
+     * })
+     * ```
      */
-    async getService(serviceFqn: string): Promise<{
-        serviceId: string
-        username: string
-        serviceFqn: string
-        offerId: string
-        sdp: string
-        createdAt: number
-        expiresAt: number
-    }> {
-        return await this.api.getService(serviceFqn)
-    }
+    async findService(
+        serviceFqn: string,
+        options?: FindServiceOptions
+    ): Promise<ServiceResult | PaginatedServiceResult> {
+        const { mode, limit = 10, offset = 0 } = options || {}
 
-    /**
-     * Discover a random available service without knowing the username
-     * Example: chat:1.0.0 (without @username)
-     */
-    async discoverService(serviceVersion: string): Promise<{
-        serviceId: string
-        username: string
-        serviceFqn: string
-        offerId: string
-        sdp: string
-        createdAt: number
-        expiresAt: number
-    }> {
-        return await this.api.getService(serviceVersion)
-    }
+        // Auto-detect mode if not specified
+        const hasUsername = serviceFqn.includes('@')
+        const effectiveMode = mode || (hasUsername ? 'direct' : 'random')
 
-    /**
-     * Discover multiple available services with pagination
-     * Example: chat:1.0.0 (without @username)
-     */
-    async discoverServices(serviceVersion: string, limit: number = 10, offset: number = 0): Promise<{
-        services: Array<{
-            serviceId: string
-            username: string
-            serviceFqn: string
-            offerId: string
-            sdp: string
-            createdAt: number
-            expiresAt: number
-        }>
-        count: number
-        limit: number
-        offset: number
-    }> {
-        return await this.api.getService(serviceVersion, { limit, offset })
+        if (effectiveMode === 'paginated') {
+            return await this.api.getService(serviceFqn, { limit, offset })
+        } else {
+            // Both 'direct' and 'random' use the same API call
+            return await this.api.getService(serviceFqn)
+        }
     }
 
     // ============================================