fix(events): fail closed when the enqueue fails

app/services/events/create_batch_service.rb

@@ -68,8 +68,10 @@ def post_validate_events
         return if result.errors.any?
       end
 
-      KafkaProducerService.call!(events: result.events, organization:)
+      # Enqueued before producing to Kafka so that a failed enqueue leaves nothing behind
+      # downstream either.
       enqueue_post_process_jobs if organization.postgres_events_store?
+      KafkaProducerService.call!(events: result.events, organization:)
     end
 
     def bulk_insert_events
@@ -100,6 +102,11 @@ def bulk_insert_events
     def enqueue_post_process_jobs
       jobs = result.events.map { |event| Events::PostProcessJob.new(event:) }
       ApplicationJob.perform_all_later(jobs)
+    rescue
+      # `perform_all_later` is a single bulk push, so one failure strands the whole batch. Hard-deleted
+      # rather than discarded for the same reason as in `Events::CreateService`.
+      Event.where(id: result.events.map(&:id)).delete_all
+      raise
     end
   end
 end

app/services/events/create_service.rb

@@ -13,14 +13,17 @@ def initialize(organization:, params:, timestamp:, metadata:)
     end
 
     def call
+      event_timestamp = parse_timestamp
+      return result.single_validation_failure!(field: :timestamp, error_code: "invalid_format") unless event_timestamp
+
       event = Event.new
       event.organization_id = organization.id
       event.code = params[:code]
       event.transaction_id = params[:transaction_id]
       event.external_subscription_id = params[:external_subscription_id]
       event.properties = params[:properties] || {}
       event.metadata = metadata || {}
-      event.timestamp = Time.zone.at(params[:timestamp] ? BigDecimal(params[:timestamp].to_s) : timestamp)
+      event.timestamp = event_timestamp
       event.precise_total_amount_cents = params[:precise_total_amount_cents]
 
       expression_result = CalculateExpressionService.call(organization:, event:)
@@ -30,24 +33,39 @@ def call
 
       result.event = event
 
+      # Enqueued before producing to Kafka so that a failed enqueue leaves nothing behind
+      # downstream either.
+      enqueue_post_process(event) unless organization.clickhouse_events_store?
       produce_kafka_event(event)
-      Events::PostProcessJob.perform_later(event:) unless organization.clickhouse_events_store?
 
       result
     rescue ActiveRecord::RecordInvalid => e
       result.record_validation_failure!(record: e.record)
     rescue ActiveRecord::RecordNotUnique
       result.single_validation_failure!(field: :transaction_id, error_code: "value_already_exist")
-    rescue ArgumentError
-      result.single_validation_failure!(field: :timestamp, error_code: "invalid_format")
     end
 
     private
 
     attr_reader :organization, :params, :timestamp, :metadata
 
+    def parse_timestamp
+      Time.zone.at(params[:timestamp] ? BigDecimal(params[:timestamp].to_s) : timestamp)
+    rescue ArgumentError
+      nil
+    end
+
     def produce_kafka_event(event)
       Events::KafkaProducerService.call!(events: event, organization:)
     end
+
+    def enqueue_post_process(event)
+      Events::PostProcessJob.perform_later(event:)
+    rescue
+      # Hard-deleted rather than discarded: `index_unique_transaction_id` carries no `deleted_at`
+      # predicate, so a discarded event would keep refusing the caller's retry.
+      event.delete
+      raise
+    end
   end
 end

spec/services/events/create_batch_service_spec.rb

@@ -89,6 +89,26 @@ def test_validation_failure(expected_errors)
       end
     end
 
+    context "when the post processing jobs cannot be enqueued" do
+      before do
+        allow(ActiveJob::Base.queue_adapter).to receive(:enqueue)
+          .and_raise(Redis::CannotConnectError.new("no connection"))
+      end
+
+      it "does not keep any of the events" do
+        expect { create_batch_service.call }.to raise_error(Redis::CannotConnectError)
+          .and(not_change(Event, :count))
+      end
+
+      it "does not produce the events on kafka" do
+        allow(Events::KafkaProducerService).to receive(:call!).and_call_original
+
+        expect { create_batch_service.call }.to raise_error(Redis::CannotConnectError)
+
+        expect(Events::KafkaProducerService).not_to have_received(:call!)
+      end
+    end
+
     context "when no events are provided" do
       let(:events_params) { build_params(count: 0) }
 

spec/services/events/create_service_spec.rb

@@ -56,6 +56,47 @@
       expect { create_service.call }.to have_enqueued_job(Events::PostProcessJob)
     end
 
+    context "when the post processing job cannot be enqueued" do
+      before do
+        allow(ActiveJob::Base.queue_adapter).to receive(:enqueue)
+          .and_raise(Redis::CannotConnectError.new("no connection"))
+      end
+
+      # `index_unique_transaction_id` has no `deleted_at` predicate, so keeping the event would
+      # answer the caller's retry with `value_already_exist` forever.
+      it "does not keep the event" do
+        expect { create_service.call }.to raise_error(Redis::CannotConnectError)
+          .and(not_change(Event, :count))
+      end
+
+      context "when kafka is configured", :capture_kafka_messages do
+        before do
+          ENV["LAGO_KAFKA_BOOTSTRAP_SERVERS"] = "kafka"
+          ENV["LAGO_KAFKA_RAW_EVENTS_TOPIC"] = "raw_events"
+        end
+
+        it "does not produce the event on kafka" do
+          expect { create_service.call }.to raise_error(Redis::CannotConnectError)
+
+          expect(karafka_producer).not_to have_received(:produce_many_async)
+        end
+      end
+    end
+
+    # `ActiveJob::SerializationError` is an `ArgumentError`, which must not be reported to the
+    # caller as an invalid timestamp.
+    context "when the job cannot be serialized" do
+      before do
+        allow(ActiveJob::Base.queue_adapter).to receive(:enqueue)
+          .and_raise(ActiveJob::SerializationError.new("unserializable"))
+      end
+
+      it "surfaces the failure instead of blaming the payload" do
+        expect { create_service.call }.to raise_error(ActiveJob::SerializationError)
+          .and(not_change(Event, :count))
+      end
+    end
+
     context "when event already exists" do
       let(:existing_event) do
         create(