single_server_queueing_stability

import Mathlib.MeasureTheory.Integral.DominatedConvergence

open Filter MeasureTheory Set
open scoped Topology

namespace SingleServerQueueing

/-- Aggregate service work arriving to a continuous-time, unit-rate, infinite-buffer
single-server queue. `workArrived s t ω` is the total service requirement arriving in `(s,t]`
on sample path `ω`. Finite real values encode local finiteness of work on bounded intervals. -/
structure Input (Ω : Type*) [MeasurableSpace Ω] (μ : Measure Ω) where
  workArrived : ℝ → ℝ → Ω → ℝ
  measurable_workArrived : ∀ s t, Measurable (workArrived s t)
  workArrived_nonneg : ∀ ω s t, s ≤ t → 0 ≤ workArrived s t ω
  workArrived_self : ∀ ω t, workArrived t t ω = 0
  workArrived_add : ∀ ω s u t, s ≤ u → u ≤ t →
    workArrived s t ω = workArrived s u ω + workArrived u t ω
  /-- Long-run offered work per unit time. Under stationary ergodic input this is almost surely
  constant and equals the expected work arriving in a unit interval. -/
  load : ℝ
  load_nonneg : 0 ≤ load
  load_tendsto_past : ∀ᵐ ω ∂μ,
    Tendsto (fun r : ℝ ↦ workArrived (-r) 0 ω / r) atTop (nhds load)
  load_tendsto_future : ∀ᵐ ω ∂μ,
    Tendsto (fun r : ℝ ↦ workArrived 0 r ω / r) atTop (nhds load)

/-- The reflection formula for workload at time `t ≥ 0`, starting with finite work `α` at
time zero. The first term represents work descended from the initial condition; the supremum
represents every possible busy period beginning after time zero. -/
noncomputable def Input.transientWorkload { Ω : Type* } [MeasurableSpace Ω] { μ : Measure Ω }
    (q : Input Ω μ) (α t : ℝ) (ω : Ω) : ℝ :=
  if 0 ≤ t then
    max (α + q.workArrived 0 t ω - t)
      (sSup {y : ℝ | ∃ u : ℝ, 0 ≤ u ∧ u ≤ t ∧
        y = q.workArrived u t ω - (t - u)})
  else α

/-- The stationary (infinite-past) workload: the greatest excess of work arriving after a
possible busy-period start `s ≤ t` over the unit-rate service available between `s` and `t`. -/
def Input.pastWorkloadCandidates { Ω : Type* } [MeasurableSpace Ω] { μ : Measure Ω }
    (q : Input Ω μ) (t : ℝ) (ω : Ω) : Set ℝ :=
  {y : ℝ | ∃ s : ℝ, s ≤ t ∧ y = q.workArrived s t ω - (t - s)}

/-- The stationary (infinite-past) workload: the greatest excess of work arriving after a
possible busy-period start `s ≤ t` over the unit-rate service available between `s` and `t`. -/
noncomputable def Input.stationaryWorkload { Ω : Type* } [MeasurableSpace Ω] { μ : Measure Ω }
    (q : Input Ω μ) (t : ℝ) (ω : Ω) : ℝ :=
  sSup (q.pastWorkloadCandidates t ω)

/-- The workload observables required below are measurable. This is explicit because the
stationary workload is an uncountable supremum; standard point-process regularity assumptions
imply it, but proving that fact is logically separate from the stability argument. -/
def Input.WorkloadsMeasurable { Ω : Type* } [MeasurableSpace Ω] { μ : Measure Ω }
    (q : Input Ω μ) : Prop :=
  (∀ α t, Measurable (q.transientWorkload α t)) ∧
    ∀ t, Measurable (q.stationaryWorkload t)

/-- Two-time stationarity of a real-valued process. This is exactly the finite-dimensional
stationarity needed for the two-time convergence theorem. -/
def IsTwoPointStationary { Ω : Type* } [MeasurableSpace Ω] (μ : Measure Ω)
    (X : ℝ → Ω → ℝ) : Prop :=
  ∀ (s₁ s₂ τ : ℝ) (B₁ B₂ : Set ℝ),
    MeasurableSet B₁ → MeasurableSet B₂ →
    μ.real {ω | X (s₁ + τ) ω ∈ B₁ ∧ X (s₂ + τ) ω ∈ B₂} =
      μ.real {ω | X s₁ ω ∈ B₁ ∧ X s₂ ω ∈ B₂}

/-- The stationary workload distribution, namely the law of the stationary workload at time
zero. Under `WorkloadsMeasurable`, this is the genuine pushforward probability distribution. -/
noncomputable def Input.stationaryWorkloadDistribution
    { Ω : Type* } [MeasurableSpace Ω] { μ : Measure Ω } (q : Input Ω μ) : Measure ℝ :=
  Measure.map (q.stationaryWorkload 0) μ

end SingleServerQueueing

• Endorsed by Community (Bot) · Jul 12, 2026

• Endorsed by wenxinzhang · Jul 12, 2026

  Confirmed by the mission captain (proposal self-audit).