tests billing engine v2

Author: brunomiguelpinto

app/jobs/billing_cycles/finalize_invoice_job.rb

@@ -7,6 +7,12 @@ class ProcessJob < ApplicationJob
     # lock_ttl auto-expires the lock so a crashed job never blocks that customer forever.
     unique :until_executed, on_conflict: :log, lock_ttl: 12.hours
 
+    # Invoice finalization (numbering) happens inline in ProcessService and serialises per
+    # billing_entity; under contention it raises SequenceError, which we retry (mirrors the
+    # legacy BillSubscriptionJob). ProcessService's reconcile makes the retry idempotent.
+    retry_on Customers::FailedToAcquireLock, ActiveRecord::StaleObjectError, attempts: MAX_LOCK_RETRY_ATTEMPTS, wait: random_lock_retry_delay
+    retry_on Sequenced::SequenceError, wait: :polynomially_longer, attempts: 15, jitter: 0.75
+
     queue_as do
       if ActiveModel::Type::Boolean.new.cast(ENV["SIDEKIQ_BILLING"])
         :billing

app/jobs/billing_cycles/process_job.rb

@@ -7,7 +7,9 @@ module Clock
   # page. The chain spreads the scan across small runs; the per-customer jobs run in
   # parallel. Deduped per customer downstream, so overlapping ticks don't double-schedule.
   class CreateBillingCyclesJob < ClockJob
-    unique :until_executed, on_conflict: :log
+    # Self-chaining, so a crashed head could otherwise hold the uniqueness lock and stall
+    # the outbox forever; lock_ttl auto-expires it well within a few 5-minute ticks.
+    unique :until_executed, on_conflict: :log, lock_ttl: 10.minutes
 
     BATCH_SIZE = 1_000
 

app/jobs/clock/create_billing_cycles_job.rb

@@ -7,7 +7,9 @@ module Clock
   # pending set into invoices. Decoupled from production; a failed cycle stays pending and
   # is retried on the next scan.
   class ProcessBillingCyclesJob < ClockJob
-    unique :until_executed, on_conflict: :log
+    # Self-chaining, so a crashed head could otherwise hold the uniqueness lock and stall
+    # the outbox forever; lock_ttl auto-expires it well within a few 5-minute ticks.
+    unique :until_executed, on_conflict: :log, lock_ttl: 10.minutes
 
     BATCH_SIZE = 1_000
 

app/jobs/clock/process_billing_cycles_job.rb

@@ -27,6 +27,15 @@ def call
         pending_cycles.group_by { |cycle| invoice_key(cycle) }.each_value do |cycles|
           result.invoices << build_invoice(cycles)
         end
+
+        # Finalize inline (invoice numbering) in the same job — one job per invoice, like
+        # the legacy BillSubscriptionJob, so no extra Sidekiq hop. Each finalize runs in
+        # its OWN short transaction (invoice.finalized!), so the per-billing_entity
+        # numbering lock is held briefly and a race raises a clean SequenceError that the
+        # ProcessJob retries. Retry-safe: a failed finalize leaves the invoice `generating`,
+        # and this reconcile re-finalizes it on the retry (it re-queries generating
+        # invoices, so it doesn't re-create — the cycles are already `done`).
+        finalize_generating_invoices
       end
 
       result
@@ -108,13 +117,24 @@ def build_invoice(cycles)
         cycles.each { |cycle| cycle.update!(status: :done, invoice:) }
       end
 
-      # Finalize (invoice numbering) in a separate, retry-protected job. Numbering
-      # serialises per billing_entity on a 10s advisory lock, so keeping it OUT of this
-      # parallel per-customer transaction means the contention can't crash the fan-out —
-      # a numbering race just retries the finalize on the already-persisted invoice.
-      BillingCycles::FinalizeInvoiceJob.perform_later(invoice)
-
       invoice
     end
+
+    # Finalize every still-generating invoice this customer's cycles produced — this run's
+    # plus any left generating by a failed finalize on a previous attempt. Re-querying
+    # (not the built array) is what makes a job retry recover orphans without re-creating.
+    def finalize_generating_invoices
+      invoice_ids = BillingCycle
+        .where(customer_id: customer.id, status: :done)
+        .where.not(invoice_id: nil)
+        .joins(:invoice)
+        .where(invoices: {status: :generating})
+        .distinct
+        .pluck(:invoice_id)
+
+      Invoice.where(id: invoice_ids).find_each do |invoice|
+        Invoices::TransitionToFinalStatusService.call(invoice:)
+      end
+    end
   end
 end

app/services/billing_cycles/process_service.rb

@@ -34,6 +34,10 @@ def call
           subscription.mark_as_terminated!(terminated_at)
         end
 
+        # Bill the final (arrears) cycle right away instead of waiting for the periodic
+        # clock — the final invoice should land on termination, matching the legacy engine.
+        after_commit { BillingCycles::BillSubscriptionJob.perform_later(subscription) }
+
         result.subscription = subscription
         result
       end

app/services/v2/subscriptions/terminate_service.rb

@@ -30,6 +30,21 @@ module Clockwork
       .perform_later
   end
 
+  # New billing engine outbox: producer schedules due cycles, consumer drains them into
+  # invoices. Both are cheap keyset scans that fan out per-customer jobs, so they run
+  # frequently for low billing latency.
+  every(5.minutes, "schedule:create_billing_cycles") do
+    Clock::CreateBillingCyclesJob
+      .set(sentry: {"slug" => "lago_create_billing_cycles", "cron" => "*/5 * * * *"})
+      .perform_later
+  end
+
+  every(5.minutes, "schedule:process_billing_cycles") do
+    Clock::ProcessBillingCyclesJob
+      .set(sentry: {"slug" => "lago_process_billing_cycles", "cron" => "*/5 * * * *"})
+      .perform_later
+  end
+
   subscription_activity_processing_interval = ENV["LAGO_SUBSCRIPTION_ACTIVITY_PROCESSING_INTERVAL_SECONDS"].presence || 1.minute
   every(subscription_activity_processing_interval.to_i.seconds, "schedule:process_subscription_activity") do
     Clock::ProcessAllSubscriptionActivitiesJob