feat(usage): make bucket serving observable

## Context

The served path delegates to the events store silently by design, so a lagging or broken
pipeline is indistinguishable from usage merely being slow, which is what usage looked like
before the buckets existed. Aggregation is the cost, so the size of the win equals coverage:
one delegated high-traffic charge keeps a subscription slow however well the rest of the plan
is served.

## Description

Every current usage lookup is now counted as served or delegated and tagged with the reason,
emitted from the event store provider, which is the only place that knows both the decision
and why it was taken. The reasons partition the declines, so a delegation rate can be
explained without reading the code, and the residue left after subtracting the ineligible
charges is the signal that something is wrong.

Alongside it, the freshness of the buckets behind a served read. It is the most recent bucket
write rather than the oldest: the oldest reports the age of the billing period, since a bucket
closed on day one is never rewritten. It cannot tell a stalled pipeline from a charge nobody
sends events for, so liveness stays with the pipeline's own loopback.

The reasons are deliberately coarse. One label per branch would mirror the code rather than
answer a question, and the cases worth separating are the ones an operator would act on
differently. A ClickHouse read that failed is not among them: it already reaches Sentry, so it
shares a label with the callers that decline the prefetch outright.

Metrics are emitted only for organizations the gate is on for, or the disabled buckets would
drown the ratio, and no metric carries an organization, subscription or charge id, so the
series count stays a function of the reason list rather than of the customer base.

The alert rules and the pipeline's own liveness loopback live outside this repository.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: vincent-pochet

app/services/events/stores/provider.rb

@@ -5,6 +5,9 @@ module Stores
     # Mints the event store instances used by a single usage or billing computation, and
     # holds the pre-aggregated usage buckets it may serve count and sum from.
     #
+    # It is also the only place that knows whether a (charge, filter) was served or delegated
+    # and why, so it is the only place that reports it.
+    #
     # Scoped to one computation at one window: build one per request or job.
     class Provider
       def initialize(organization:, subscription:, boundaries:, usage_buckets: nil, current_usage: false)
@@ -13,6 +16,7 @@ def initialize(organization:, subscription:, boundaries:, usage_buckets: nil, cu
         @boundaries = boundaries
         @usage_buckets = usage_buckets
         @current_usage = current_usage
+        @outcomes = {}
       end
 
       attr_reader :subscription, :boundaries
@@ -74,27 +78,16 @@ def precomputed_options_for(charge:, filters: {})
         }
       end
 
-      # Whether this (charge, filter) is answered from the buckets. The totals answer for the
-      # whole (charge, filter), so a group-scoped or pay-in-advance read cannot use them, and
-      # a presentation breakdown reads events anyway. A window without a single bucket is a
-      # pipeline gap rather than an absence of usage: answering zero would undercharge.
-      #
-      # The buckets close 15 minutes at a time, so they always lag: current usage can read a
-      # lagging total, an invoice cannot. A `max_timestamp` freezes the read below the window
-      # the totals cover, which would overcount by everything that landed after it.
+      # Whether this (charge, filter) is answered from the buckets. Both callers of the decision
+      # — the charge cache bypass and the aggregation options — come through here, so the
+      # outcome is memoized and a lookup is reported once.
       def serves?(charge:, filters: {})
-        return false if usage_buckets.blank?
         return false unless current_usage
-        return false if boundaries[:max_timestamp].present?
-        return false unless RealtimeUsage.enabled?(organization)
-        return false if RealtimeUsage.deduplicated?(organization)
-        return false unless RealtimeUsage.supported_charge?(charge)
-        return false unless usage_buckets.serves_charge?(charge.id)
-
-        filters[:grouped_by_values].blank? &&
-          filters[:event].blank? &&
-          filters[:presentation_by].blank? &&
-          filters[:filter_by_group].blank?
+
+        key = [charge.id, self.class.bucket_charge_filter_id(filters[:charge_filter])]
+        return outcomes[key] if outcomes.key?(key)
+
+        outcomes[key] = report(delegation_reason(charge:, filters:))
       end
 
       # The sink writes `COALESCE(charge_filter_id, '')`, while the unfiltered fee carries
@@ -105,7 +98,74 @@ def self.bucket_charge_filter_id(charge_filter)
 
       private
 
-      attr_reader :organization, :current_usage
+      attr_reader :organization, :current_usage, :outcomes
+
+      # Asked once per (charge, filter) otherwise, on the path the buckets exist to make fast.
+      def gate_open?
+        return @gate_open if defined?(@gate_open)
+
+        @gate_open = RealtimeUsage.enabled?(organization)
+      end
+
+      # nil when the buckets answer. The reason returned is the first thing that would have to
+      # change for this lookup to be served, so a plan of ineligible charges reports what makes
+      # them ineligible rather than the absent prefetch that ineligibility caused.
+      #
+      # The buckets close 15 minutes at a time, so they always lag: current usage can read a
+      # lagging total, an invoice cannot. A `max_timestamp` freezes the read below the window
+      # the totals cover, which would overcount by everything that landed after it.
+      #
+      # A window without a single bucket is a pipeline gap rather than an absence of usage:
+      # answering zero would undercharge.
+      #
+      # `not_prefetched` covers both a caller that declined the prefetch — `full_usage` and
+      # projected reads do — and a ClickHouse read that failed, which already reaches Sentry.
+      def delegation_reason(charge:, filters:)
+        return :gate_disabled unless gate_open?
+        return :deduplicated if RealtimeUsage.deduplicated?(organization)
+        return :frozen_window if boundaries[:max_timestamp].present?
+        return :ineligible_charge unless RealtimeUsage.supported_charge?(charge)
+        return :unsupported_read if unsupported_read?(filters)
+        return :not_prefetched if usage_buckets.nil?
+        return :no_buckets if usage_buckets.empty?
+
+        :drift unless usage_buckets.serves_charge?(charge.id)
+      end
+
+      # The totals answer for the whole (charge, filter), so a group-scoped or pay-in-advance
+      # read cannot use them, and a presentation breakdown reads events anyway.
+      def unsupported_read?(filters)
+        filters[:grouped_by_values].present? ||
+          filters[:event].present? ||
+          filters[:presentation_by].present? ||
+          filters[:filter_by_group].present?
+      end
+
+      # Nothing is reported for an organization the gate is shut for, or the disabled buckets
+      # would drown the ratio.
+      def report(reason)
+        served = reason.nil?
+        return served unless gate_open?
+
+        Yabeda.realtime_usage.lookups_total.increment(
+          {outcome: served ? "served" : "delegated", reason: reason&.to_s || "none"}
+        )
+        report_freshness if served
+
+        served
+      end
+
+      # Once per computation: the watermark answers for the whole prefetched set, not for the
+      # charge that happened to be looked up first.
+      def report_freshness
+        return if @freshness_reported
+
+        @freshness_reported = true
+        ingested_at = usage_buckets.last_ingested_at
+        return if ingested_at.nil?
+
+        Yabeda.realtime_usage.freshness.measure({}, (Time.current - ingested_at).to_f)
+      end
     end
   end
 end

app/services/events/stores/usage_bucket_set.rb

@@ -9,15 +9,20 @@ class UsageBucketSet
 
       # Copied before freezing: the caller usually builds these hashes as accumulators, and
       # freezing its own object would raise on the next write, far from here.
-      def initialize(totals: {}, grouped_totals: {}, unservable_charge_ids: [])
+      def initialize(totals: {}, grouped_totals: {}, unservable_charge_ids: [], last_ingested_at: nil)
         @totals = totals.dup.freeze
         @grouped_totals = grouped_totals.dup.freeze
         @unservable_charge_ids = unservable_charge_ids.to_set.freeze
+        @last_ingested_at = last_ingested_at
         freeze
       end
 
       attr_reader :unservable_charge_ids
 
+      # The watermark the whole set answers for: nil when it holds no row, so a computation
+      # served entirely from absent rows reports no freshness rather than an age of forever.
+      attr_reader :last_ingested_at
+
       def empty?
         totals.empty? && grouped_totals.empty?
       end

app/services/realtime_usage/fetch_buckets_service.rb

@@ -44,7 +44,7 @@ def organization
     # An unreachable ClickHouse has to make current usage slow, not broken: no set at all
     # leaves every charge reading events.
     def fetch
-      Events::Stores::UsageBucketSet.new(totals:, grouped_totals:, unservable_charge_ids:)
+      Events::Stores::UsageBucketSet.new(totals:, grouped_totals:, unservable_charge_ids:, last_ingested_at:)
     rescue *READ_ERRORS => e
       Rails.logger.warn("Realtime usage bucket prefetch failed: #{e.class} #{e.message}")
       Sentry.capture_exception(e)
@@ -77,15 +77,25 @@ def sum_totals(totals, row)
       )
     end
 
+    # When the sink last wrote anything into this window. Taking the oldest instead would report
+    # the age of the billing period: a bucket closed on day one is never rewritten.
+    #
+    # It cannot tell a stalled pipeline from a charge nobody sends events for, which is why
+    # liveness comes from the pipeline's own loopback rather than from here.
+    def last_ingested_at
+      rows.filter_map { it[:last_ingested_at] }.max
+    end
+
     def rows
-      @rows ||= read_rows.map do |charge_id, charge_filter_id, grouped_by, aggregation_type, units, events_count|
+      @rows ||= read_rows.map do |charge_id, charge_filter_id, grouped_by, aggregation_type, units, events_count, last_ingested_at|
         {
           charge_id:,
           charge_filter_id:,
           aggregation_type:,
           groups: parse_groups(grouped_by),
           units: units.to_d,
-          events_count: events_count.to_i
+          events_count: events_count.to_i,
+          last_ingested_at: parse_time(last_ingested_at)
         }
       end
     end
@@ -100,7 +110,7 @@ def read_rows
           .where(organization_id: organization.id, subscription_id: subscription.id, charge_id: charge_ids)
           .where(bucket: window, is_deleted: 0)
           .group(:charge_id, :charge_filter_id, :grouped_by, :aggregation_type)
-          .pluck(Arel.sql("charge_id, charge_filter_id, grouped_by, aggregation_type, sum(units), sum(events_count)"))
+          .pluck(Arel.sql("charge_id, charge_filter_id, grouped_by, aggregation_type, sum(units), sum(events_count), max(last_ingested_at)"))
       end
     end
 
@@ -120,6 +130,12 @@ def parse_groups(grouped_by)
       nil
     end
 
+    def parse_time(value)
+      return nil if value.blank?
+
+      value.is_a?(String) ? Time.zone.parse(value) : value.to_time
+    end
+
     # Rails asks the store for one group per key of `Fees::ChargeService#grouped_by_keys`, so a
     # row grouped or aggregated another way answers another question than the fee asks. Only the
     # charge as a whole sees it, hence one drifting row delegating every filter of the charge.

config/initializers/yabeda.rb

@@ -14,4 +14,20 @@
   default_tag :service, ENV["OTEL_SERVICE_NAME"] || "lago-api"
   default_tag :environment, Rails.env
   default_tag :version, ENV["LAGO_VERSION"] || "unknown"
+
+  # Emitted only for organizations the realtime usage gate is on for, and only by
+  # Events::Stores::Provider. No organization, subscription or charge id is ever a tag: the
+  # series count has to stay a function of the reason list, not of the customer base.
+  group :realtime_usage do
+    counter :lookups_total,
+      comment: "Current usage lookups answered from the pre-aggregated buckets or delegated to the events store",
+      tags: %i[outcome reason]
+
+    # The prometheus adapter builds the exported name as group_name_unit, so the metric is named
+    # without the suffix its unit already adds.
+    histogram :freshness,
+      comment: "Age of the most recent bucket write backing a served computation",
+      unit: :seconds,
+      buckets: [30, 60, 120, 300, 600, 1800, 3600, 21_600]
+  end
 end

spec/services/events/stores/provider_spec.rb

@@ -345,6 +345,165 @@
     end
   end
 
+  describe "the reported outcome" do
+    subject(:provider) do
+      described_class.new(organization:, subscription:, boundaries:, usage_buckets: bucket_set, current_usage: true)
+    end
+
+    include_context "with realtime usage availability"
+
+    let(:organization) do
+      create(:organization, clickhouse_events_store: true, feature_flags: ["realtime_usage"])
+    end
+    let(:reported) { [] }
+
+    before { allow(Yabeda.realtime_usage.lookups_total).to receive(:increment) { |tags| reported << tags } }
+
+    it "reports a served lookup once, however many callers ask the same question" do
+      provider.serves?(charge:)
+      provider.serves?(charge:)
+
+      expect(reported).to eq([{outcome: "served", reason: "none"}])
+    end
+
+    it "reports each filter of a charge separately, as each is a lookup a fee took" do
+      provider.serves?(charge:)
+      provider.serves?(charge:, filters: {charge_filter: create(:charge_filter, charge:)})
+
+      expect(reported.size).to eq(2)
+    end
+
+    it "reports nothing outside a current usage computation, which never asked for buckets" do
+      described_class.new(organization:, subscription:, boundaries:, usage_buckets: bucket_set).serves?(charge:)
+
+      expect(reported).to be_empty
+    end
+
+    it "reports nothing for an organization the gate is shut for, whose declines would drown the ratio" do
+      other_organization = create(:organization, clickhouse_events_store: true)
+
+      described_class
+        .new(organization: other_organization, subscription:, boundaries:, usage_buckets: bucket_set, current_usage: true)
+        .serves?(charge:)
+
+      expect(reported).to be_empty
+    end
+
+    context "when the organization deduplicates its events" do
+      let(:organization) do
+        create(:organization, clickhouse_events_store: true, feature_flags: ["realtime_usage"], clickhouse_deduplication_enabled: true)
+      end
+
+      it "reports the delegation" do
+        provider.serves?(charge:)
+
+        expect(reported).to eq([{outcome: "delegated", reason: "deduplicated"}])
+      end
+    end
+
+    context "when the read is frozen at a past timestamp" do
+      let(:boundaries) { {from_datetime: Time.current.beginning_of_month, to_datetime: Time.current, max_timestamp: 1.day.ago} }
+
+      it "reports the delegation" do
+        provider.serves?(charge:)
+
+        expect(reported).to eq([{outcome: "delegated", reason: "frozen_window"}])
+      end
+    end
+
+    context "with a charge the buckets were never going to answer" do
+      let(:charge) { create(:percentage_charge, plan: subscription.plan, billable_metric:) }
+
+      it "reports the delegation" do
+        provider.serves?(charge:)
+
+        expect(reported).to eq([{outcome: "delegated", reason: "ineligible_charge"}])
+      end
+    end
+
+    it "reports a read whose shape the totals cannot answer" do
+      provider.serves?(charge:, filters: {presentation_by: ["region"]})
+
+      expect(reported).to eq([{outcome: "delegated", reason: "unsupported_read"}])
+    end
+
+    context "when the computation holds no prefetch" do
+      let(:bucket_set) { nil }
+
+      it "reports the delegation" do
+        provider.serves?(charge:)
+
+        expect(reported).to eq([{outcome: "delegated", reason: "not_prefetched"}])
+      end
+    end
+
+    context "when the window holds no bucket" do
+      let(:bucket_set) { Events::Stores::UsageBucketSet.new }
+
+      it "reports the delegation, as answering zero would undercharge" do
+        provider.serves?(charge:)
+
+        expect(reported).to eq([{outcome: "delegated", reason: "no_buckets"}])
+      end
+    end
+
+    context "with a charge the prefetch found drifting" do
+      let(:bucket_set) do
+        Events::Stores::UsageBucketSet.new(
+          totals: {[charge.id, ""] => Events::Stores::UsageBucketSet::Totals.new(units: BigDecimal("10"), events_count: 2)},
+          unservable_charge_ids: [charge.id]
+        )
+      end
+
+      it "reports the delegation" do
+        provider.serves?(charge:)
+
+        expect(reported).to eq([{outcome: "delegated", reason: "drift"}])
+      end
+    end
+  end
+
+  describe "the reported freshness" do
+    subject(:provider) do
+      described_class.new(organization:, subscription:, boundaries:, usage_buckets: bucket_set, current_usage: true)
+    end
+
+    include_context "with realtime usage availability"
+
+    let(:organization) do
+      create(:organization, clickhouse_events_store: true, feature_flags: ["realtime_usage"])
+    end
+    let(:bucket_set) do
+      Events::Stores::UsageBucketSet.new(
+        totals: {[charge.id, ""] => Events::Stores::UsageBucketSet::Totals.new(units: BigDecimal("10"), events_count: 2)},
+        last_ingested_at: 2.minutes.ago
+      )
+    end
+
+    before { allow(Yabeda.realtime_usage.freshness).to receive(:measure) }
+
+    it "measures the age of the last bucket write once, as the watermark answers for the whole set" do
+      provider.serves?(charge:)
+      provider.serves?(charge:, filters: {charge_filter: create(:charge_filter, charge:)})
+
+      expect(Yabeda.realtime_usage.freshness).to have_received(:measure).once.with({}, be_within(5).of(120))
+    end
+
+    context "when the set carries no watermark" do
+      let(:bucket_set) do
+        Events::Stores::UsageBucketSet.new(
+          totals: {[charge.id, ""] => Events::Stores::UsageBucketSet::Totals.new(units: BigDecimal("10"), events_count: 2)}
+        )
+      end
+
+      it "measures nothing, rather than an age of forever" do
+        provider.serves?(charge:)
+
+        expect(Yabeda.realtime_usage.freshness).not_to have_received(:measure)
+      end
+    end
+  end
+
   describe "#plain_store" do
     it "returns a store with no metric code" do
       store = provider.plain_store