fix(payments): stop stale invoices from wedging payment polling

.changeset/invoice-default-expiry.md

@@ -0,0 +1,21 @@
+---
+"nostream": patch
+---
+
+fix(payments): stop stale invoices from wedging payment polling
+
+A relay could stop clearing payments entirely, needing `delete from invoices` to
+recover. Two things combined to cause it.
+
+Invoices created without an expiry could never be retired, because the expiry
+check treats a missing date as "not expired", so the maintenance worker left them
+pending forever. The LNURL processor set no expiry on any invoice, making this
+certain there rather than incidental. Invoices now fall back to
+`payments.invoiceExpirySeconds` when the processor reports no expiry of its own,
+and existing pending rows without one are backfilled.
+
+Separately, each maintenance pass re-read the same oldest page of pending
+invoices, so one page of invoices that never resolve starved every newer one
+indefinitely. The worker now advances through the queue and wraps at the end,
+keeping the same per-pass cost while guaranteeing every pending invoice is
+eventually polled.

migrations/20260829_120000_backfill_null_invoice_expiry.js

@@ -0,0 +1,21 @@
+// isExpiredInvoice is false for a null expiry, so the worker never retires these
+// and they sit in its polling window forever. Give the existing ones an expiry
+// from their creation time so they can drain. Pending rows only.
+const DEFAULT_INVOICE_EXPIRY_SECONDS = 86400
+
+exports.up = async function (knex) {
+  await knex('invoices')
+    .whereNull('expires_at')
+    .andWhere('status', 'pending')
+    .update({
+      // created_at is timestamptz, expires_at is not. Pin the conversion to UTC
+      // instead of the session TimeZone.
+      expires_at: knex.raw("(created_at AT TIME ZONE 'UTC') + (? || ' seconds')::interval", [
+        DEFAULT_INVOICE_EXPIRY_SECONDS,
+      ]),
+    })
+}
+
+exports.down = async function () {
+  // Not reversible: a backfilled expiry is indistinguishable from a real one.
+}

resources/default-settings.yaml

@@ -21,6 +21,12 @@ payments:
             - replace-with-your-pubkey-in-hex
           event_kinds:
             - 9735 # Nip-57 Lightning Zap Receipts
+  # Applied only when the payments processor does not report an expiry of its own.
+  # Without it such invoices can never be retired and stay pending forever.
+  # Raise it if your processor issues invoices that stay payable for longer than
+  # this: LNURL and NWC have no callback, so a payment made after the invoice has
+  # been retired would not be noticed. Capped at 30 days.
+  invoiceExpirySeconds: 86400
 paymentsProcessors:
   zebedee:
     baseURL: https://api.zebedee.io/

src/@types/services.ts

@@ -12,5 +12,5 @@ export interface IPaymentsService {
   updateInvoiceStatus(invoice: Pick<Invoice, 'id' | 'status'>): Promise<Invoice>
   confirmInvoice(invoice: Pick<Invoice, 'id' | 'amountPaid' | 'confirmedAt' | 'status' | 'pubkey'>): Promise<void>
   sendInvoiceUpdateNotification(invoice: Invoice): Promise<void>
-  getPendingInvoices(): Promise<Invoice[]>
+  getPendingInvoices(offset?: number): Promise<Invoice[]>
 }

src/@types/settings.ts

@@ -183,6 +183,8 @@ export interface Payments {
   enabled: boolean
   processor: keyof PaymentsProcessors
   feeSchedules: FeeSchedules
+  /** Fallback when the processor reports no expiry. A reported one always wins. */
+  invoiceExpirySeconds?: number
 }
 
 export interface LnurlPaymentsProcessor {

src/app/maintenance-worker.ts

@@ -9,6 +9,7 @@ import { mergeDeepLeft, path, pipe } from 'ramda'
 import { IRunnable } from '../@types/base'
 
 import { createLogger } from '../factories/logger-factory'
+import { PENDING_INVOICE_PAGE_SIZE } from '../services/payments-service'
 import { delayMs } from '../utils/misc'
 import { INip05VerificationRepository } from '../@types/repositories'
 import { InvoiceStatus } from '../@types/invoice'
@@ -23,8 +24,7 @@ const CLEAR_OLD_EVENTS_TIMEOUT_MS = 5000
 
 const logger = createLogger('maintenance-worker')
 
-const isNotFoundError = (error: unknown): boolean =>
-  (error as any)?.response?.status === 404
+const isNotFoundError = (error: unknown): boolean => (error as any)?.response?.status === 404
 
 /**
  * Merge a re-verification outcome onto an existing verification row.
@@ -74,6 +74,11 @@ export function applyReverificationOutcome(
 export class MaintenanceWorker implements IRunnable {
   private interval: NodeJS.Timeout | undefined
   private isRunning = false
+  /**
+   * Where the next pass starts. Without it every pass re-reads the oldest ten, so
+   * ten invoices that never resolve starve everything behind them.
+   */
+  private pendingInvoiceOffset = 0
 
   public constructor(
     private readonly process: NodeJS.Process,
@@ -132,8 +137,12 @@ export class MaintenanceWorker implements IRunnable {
       return
     }
 
-    const invoices = await this.paymentsService.getPendingInvoices()
-    logger('found %d pending invoices', invoices.length)
+    const invoices = await this.paymentsService.getPendingInvoices(this.pendingInvoiceOffset)
+    logger('found %d pending invoices from offset %d', invoices.length, this.pendingInvoiceOffset)
+
+    // A short page means we reached the end, so start over next pass.
+    this.pendingInvoiceOffset =
+      invoices.length < PENDING_INVOICE_PAGE_SIZE ? 0 : this.pendingInvoiceOffset + PENDING_INVOICE_PAGE_SIZE
     const delay = () => delayMs(100 + Math.floor(Math.random() * 10))
 
     let successful = 0

src/services/payments-service.ts

@@ -11,9 +11,13 @@ import { createLogger } from '../factories/logger-factory'
 import { IPaymentsProcessor } from '../@types/clients'
 import { IPaymentsService } from '../@types/services'
 import { Transaction } from '../database/transaction'
+import { resolveInvoiceExpiry } from '../utils/invoice'
 
 const logger = createLogger('payments-service')
 
+/** Invoices per maintenance pass. Small because each one is a processor round trip. */
+export const PENDING_INVOICE_PAGE_SIZE = 10
+
 export class PaymentsService implements IPaymentsService {
   public constructor(
     private readonly dbClient: DatabaseClient,
@@ -24,10 +28,10 @@ export class PaymentsService implements IPaymentsService {
     private readonly settings: () => Settings,
   ) {}
 
-  public async getPendingInvoices(): Promise<Invoice[]> {
-    logger('get pending invoices')
+  public async getPendingInvoices(offset = 0): Promise<Invoice[]> {
+    logger('get pending invoices from offset %d', offset)
     try {
-      return await this.invoiceRepository.findPendingInvoices(0, 10)
+      return await this.invoiceRepository.findPendingInvoices(offset, PENDING_INVOICE_PAGE_SIZE)
     } catch (error) {
       logger.error('Unable to get pending invoices.', error)
 
@@ -63,6 +67,11 @@ export class PaymentsService implements IPaymentsService {
       })
 
       const date = new Date()
+      const expiresAt = resolveInvoiceExpiry(
+        invoiceResponse.expiresAt,
+        date,
+        this.settings()?.payments?.invoiceExpirySeconds,
+      )
 
       await this.invoiceRepository.upsert(
         {
@@ -73,7 +82,7 @@ export class PaymentsService implements IPaymentsService {
           description: invoiceResponse.description,
           unit: invoiceResponse.unit,
           status: invoiceResponse.status,
-          expiresAt: invoiceResponse.expiresAt,
+          expiresAt,
           updatedAt: date,
           createdAt: date,
           verifyURL: invoiceResponse.verifyURL,
@@ -91,7 +100,7 @@ export class PaymentsService implements IPaymentsService {
         unit: invoiceResponse.unit,
         status: invoiceResponse.status,
         description,
-        expiresAt: invoiceResponse.expiresAt,
+        expiresAt,
         updatedAt: date,
         createdAt: invoiceResponse.createdAt,
         verifyURL: invoiceResponse.verifyURL,

src/utils/invoice.ts

@@ -1,2 +1,40 @@
+/**
+ * Fallback when the processor reports no expiry. Generous on purpose: LNURL and NWC
+ * have no callback, so retiring an invoice that is still payable loses the payment,
+ * while erring long just leaves the row around a bit.
+ */
+export const DEFAULT_INVOICE_EXPIRY_SECONDS = 86400
+
+/** Cap on the configured fallback. An over-long value is the bug this prevents. */
+export const MAX_INVOICE_EXPIRY_SECONDS = 30 * 86400
+
 export const isExpiredInvoice = (invoice: { expiresAt?: Date | null }): boolean =>
   invoice.expiresAt instanceof Date && invoice.expiresAt.getTime() <= Date.now()
+
+const isUsableDate = (value: unknown): value is Date => value instanceof Date && !Number.isNaN(value.getTime())
+
+export const resolveInvoiceExpirySeconds = (configured: unknown): number => {
+  if (typeof configured === 'number' && Number.isSafeInteger(configured) && configured > 0) {
+    return Math.min(configured, MAX_INVOICE_EXPIRY_SECONDS)
+  }
+
+  return DEFAULT_INVOICE_EXPIRY_SECONDS
+}
+
+/**
+ * Every invoice needs an expiry. `isExpiredInvoice` is false for null and for an
+ * unparseable date, so without one the row can never be retired.
+ */
+export const resolveInvoiceExpiry = (
+  processorExpiry: Date | null | undefined,
+  createdAt: Date,
+  expirySeconds: number = DEFAULT_INVOICE_EXPIRY_SECONDS,
+): Date => {
+  if (isUsableDate(processorExpiry)) {
+    return processorExpiry
+  }
+
+  const base = isUsableDate(createdAt) ? createdAt : new Date()
+
+  return new Date(base.getTime() + resolveInvoiceExpirySeconds(expirySeconds) * 1000)
+}

test/unit/app/maintenance-worker.spec.ts

@@ -433,15 +433,73 @@ describe('MaintenanceWorker', () => {
       settingsState.payments = { enabled: true } as any
       paymentsService.getPendingInvoices.resolves([pendingInvoice, secondInvoice])
       paymentsService.getInvoiceFromPaymentsProcessor
-        .onFirstCall().rejects(new Error('processor error'))
-        .onSecondCall().resolves({ id: 'inv-2', status: InvoiceStatus.PENDING })
+        .onFirstCall()
+        .rejects(new Error('processor error'))
+        .onSecondCall()
+        .resolves({ id: 'inv-2', status: InvoiceStatus.PENDING })
 
       await (worker as any).onSchedule()
 
       expect(maintenanceService.clearOldEvents).to.have.been.calledOnce
       expect(paymentsService.updateInvoiceStatus).to.have.been.calledOnce
     })
 
+    it('walks the pending queue instead of re-reading the same page', async () => {
+      // A full page means there may be more behind it, so start further in next time.
+      settingsState.payments = { enabled: true } as any
+      const fullPage = Array.from({ length: 10 }, (_, i) => ({ ...pendingInvoice, id: `inv-${i}` }))
+      paymentsService.getPendingInvoices.resolves(fullPage)
+      paymentsService.getInvoiceFromPaymentsProcessor.resolves({
+        id: 'inv-0',
+        status: InvoiceStatus.PENDING,
+      })
+
+      await (worker as any).onSchedule()
+      await (worker as any).onSchedule()
+      await (worker as any).onSchedule()
+
+      expect(paymentsService.getPendingInvoices.getCall(0).args[0]).to.equal(0)
+      expect(paymentsService.getPendingInvoices.getCall(1).args[0]).to.equal(10)
+      expect(paymentsService.getPendingInvoices.getCall(2).args[0]).to.equal(20)
+    })
+
+    it('starts over once it reaches the end of the queue', async () => {
+      settingsState.payments = { enabled: true } as any
+      const fullPage = Array.from({ length: 10 }, (_, i) => ({ ...pendingInvoice, id: `inv-${i}` }))
+      paymentsService.getInvoiceFromPaymentsProcessor.resolves({
+        id: 'inv-0',
+        status: InvoiceStatus.PENDING,
+      })
+
+      paymentsService.getPendingInvoices.resolves(fullPage)
+      await (worker as any).onSchedule()
+
+      // Short page: nothing left behind it.
+      paymentsService.getPendingInvoices.resolves([pendingInvoice])
+      await (worker as any).onSchedule()
+
+      paymentsService.getPendingInvoices.resolves(fullPage)
+      await (worker as any).onSchedule()
+
+      expect(paymentsService.getPendingInvoices.getCall(1).args[0]).to.equal(10)
+      expect(paymentsService.getPendingInvoices.getCall(2).args[0]).to.equal(0)
+    })
+
+    it('stays at the start while there is only ever one short page', async () => {
+      settingsState.payments = { enabled: true } as any
+      paymentsService.getPendingInvoices.resolves([pendingInvoice])
+      paymentsService.getInvoiceFromPaymentsProcessor.resolves({
+        id: pendingInvoice.id,
+        status: InvoiceStatus.PENDING,
+      })
+
+      await (worker as any).onSchedule()
+      await (worker as any).onSchedule()
+
+      expect(paymentsService.getPendingInvoices.getCall(0).args[0]).to.equal(0)
+      expect(paymentsService.getPendingInvoices.getCall(1).args[0]).to.equal(0)
+    })
+
     it('marks an expired pending invoice as expired when the payment processor returns 404', async () => {
       const expiredInvoice = {
         ...pendingInvoice,

test/unit/services/payments-service.spec.ts

@@ -188,6 +188,8 @@ describe('PaymentsService', () => {
     })
 
     it('upserts user, creates invoice via processor, persists, and returns the invoice', async () => {
+      settings.returns({ payments: { invoiceExpirySeconds: 3600 } })
+
       const result = await service.createInvoice('pubkey1234', 1000n, 'test')
 
       expect(dbClient.transaction).to.have.been.called
@@ -203,6 +205,44 @@ describe('PaymentsService', () => {
       expect(result.pubkey).to.equal('pubkey1234')
     })
 
+    it('gives the invoice an expiry when the processor does not report one', async () => {
+      // Without this the row can never be retired and stays pending forever.
+      settings.returns({ payments: { invoiceExpirySeconds: 3600 } })
+
+      const result = await service.createInvoice('pubkey1234', 1000n, 'test')
+
+      expect(result.expiresAt).to.be.instanceOf(Date)
+      const [persisted] = invoiceRepository.upsert.firstCall.args
+      expect(persisted.expiresAt).to.deep.equal(result.expiresAt)
+    })
+
+    it('keeps the expiry the processor reported', async () => {
+      const processorExpiry = new Date('2030-06-01T00:00:00.000Z')
+      settings.returns({ payments: { invoiceExpirySeconds: 3600 } })
+      paymentsProcessor.createInvoice.resolves({
+        id: 'new-inv-id',
+        bolt11: 'lnbc',
+        amountRequested: 1000n,
+        description: 'test',
+        unit: InvoiceUnit.MSATS,
+        status: InvoiceStatus.PENDING,
+        expiresAt: processorExpiry,
+        createdAt: new Date(),
+      })
+
+      const result = await service.createInvoice('pubkey1234', 1000n, 'test')
+
+      expect(result.expiresAt).to.equal(processorExpiry)
+    })
+
+    it('still creates the invoice when settings are unavailable', async () => {
+      settings.returns(undefined)
+
+      const result = await service.createInvoice('pubkey1234', 1000n, 'test')
+
+      expect(result.expiresAt).to.be.instanceOf(Date)
+    })
+
     it('rolls back the transaction and re-throws when the processor fails', async () => {
       paymentsProcessor.createInvoice.rejects(new Error('processor fail'))