07-EFT.WP.Core.Threads v1.0 | Chapter 3 — Message Channels and Backpressure

Home ／ Docs-Technical WhitePaper ／ 07-EFT.WP.Core.Threads v1.0

Chapter 3 — Message Channels and Backpressure

I. Scope and Objectives

• Define unified semantics and metrics for channels chan, messages msg, capacity cap, queue length q_len, backpressure bp, and acknowledgments ack.
• Provide steady-state conditions, drop policies, batching and ACK protocols, rate-limit coordination, and a closed-loop control flow Mx-2.
• Deliver postulates P73-, minimal equations S73-, and interface bindings to I70-3 (see implementation section of this volume).

II. Channel Model and Message Format

1. Channel shape
   • chan = {name, mode, cap, policy, metrics}, where mode ∈ {"spsc","mpsc","spmc","mpmc"}.
   • Queue state: q_len(t), cap, age_head(t), blocked_put/blocked_get.
2. Message minimal fields
   • msg = {eid, ts, tau_mono, idemp_key?, sem, ttl?, deadline?, payload_ref, trace_ctx?}.
   • Delivery semantics sem ∈ {"at_most_once","at_least_once","exactly_once*"} (*best-effort via deduplication).

III. Steady State and Queue Metrics (S73 Series)

1. Core definitions and identities
   • Utilization: rho = lambda / mu; steady-state necessary condition: rho < 1.
   • Little’s Law: L = lambda * W; queue form: L_q = lambda * W_q.
2. Approximate formulas (M/M/1)
   • S73-1: W_q approx rho / ( mu * (1 - rho) ).
   • S73-2: W approx 1 / ( mu - lambda ).
3. Batching correction
   With batch size b >= 1, effective service rate mu_eff approx mu * b; equivalent queue length q_len_eff = ceil( q_len / b ).

IV. Postulates and Constraints (P73-*)

• P73-1 (Capacity declaration): every chan must declare cap; cap > 0 and is constant or piecewise-constant.
• P73-2 (Stability): when using steady-state approximations, record the estimation provenance of lambda and mu, and ensure rho < 1 approx stability.
• P73-3 (Closed-loop backpressure): bp is a monotone non-decreasing function of q_len/cap and W_q; thresholds and actions are declared in pairs.
• P73-4 (Overflow visibility): upon drop or degradation, emit per-eid audit events and increment TS.drop.count.
• P73-5 (ACK time bounds): ACK modes must declare ack_timeout and max retries; satisfy W_retry <= timeout * (retries + 1) + J_total.
• P73-6 (Idempotence): when sem!="at_most_once", consumer handlers must satisfy f(x; idemp_key) = f(x; idemp_key) and declare Delta_t_dedup.

V. Backpressure Signals and Control Functions

1. Backpressure intensity
   • Define bp = min( 1 , max( 0 , a * ( q_len / cap ) + b * ( W_q / W_q_target ) ) ), with a,b ∈ [0,1].
   • Watermarks: wm_low < wm_high. When q_len/cap >= wm_high, trigger rate-limit or drop; relax when below wm_low.
2. Control actions
   • Ingress: shaping(rps), increase batch_size, priority demotion.
   • Egress: decrease prefetch, increase consumer K_thr, scale out workers.
3. Feedback cadence
   Update period T_bp should satisfy T_bp << W_q to avoid under/overshoot.

VI. Overflow and Drop Policies

1. Policy families and applicability
   • tail_drop: drop newly arrived messages; typical for ordered channels.
   • head_drop: drop oldest messages; favors timeliness (minimizes stale backlog).
   • random_drop: RED-like smoothing of congestion.
   • shed_by_key(id): rate-limit or drop by idemp_key to suppress noisy sources.
2. Selection guidelines
   • Optimize P99(latency): prefer head_drop or random_drop.
   • Optimize accuracy: prefer tail_drop or blocking (evaluate upstream rho).

VII. Batching and ACK Semantics

• Produce-side batching
  chan_put(ch, batch{msg_1..msg_b}); flush when b_target or timeout_batch is reached.
• Consume-side batching
  chan_get(ch, max_batch) returns [msg...]; acknowledge via ack(batch_id) or per-message ack(eid).
• Failure and retry
  On ack_timeout, p_retry triggers re-delivery; for sem="at_least_once", combine with dedup window Delta_t_dedup.
• Ordering and idempotence
  For order=strict, ACKs must be in-order or carry offsets; visibility respects hb.

VIII. Delivery Semantics and Deduplication

1. Semantics
   • at_most_once: no retry, p_drop can be non-zero; lowest latency.
   • at_least_once: duplicates allowed; use idemp_key for deduplication.
   • exactly_once*: approximate via idemp_key + idempotent storage.
2. Dedup window
   Maintain dedup_table over (idemp_key, ts_last); rule: if tau_mono - ts_last <= Delta_t_dedup then drop duplicate.

IX. End-to-End Timing and Age Control

1. Age and expiry
   Message age: age(msg,t) = t - ts; expiry when age(msg,t) > ttl.
2. Delay budget decomposition
   budget_total = W_q_in + w(proc) + W_q_out; target P99(budget_total) <= SLO.
3. Arrival-time reference
   If using propagation arrival time T_arr to calibrate cross-domain timing, adopt both forms:
   • T_arr = ( 1 / c_ref ) * ( ∫ n_eff d ell )
   • T_arr = ( ∫ ( n_eff / c_ref ) d ell )
   • Report the discrepancy delta_form.

X. Rate-Limit Coordination and Stability Conditions

• Rate limiter
  Token bucket: rate_limiter(name, rps, burst); consumers call limit_acquire.
• Stability review
  If bp≈1 for > T_hold, reduce rps := max( rps_min , rps * k_down ); if bp<wm_low for > T_relax, increase rps := min( rps_max , rps * k_up ) (AIMD).
• Multi-stage chains
  For a chain of channels, the bottleneck mu* dictates global rho*: rho* = max_k ( lambda_k / mu_k ).

XI. Capacity Planning and Baselines

• Baseline sizing
  cap >= ceil( lambda_peak * T_absorb ), where T_absorb is the jitter absorption window.
• Initial targets
  wm_low ≈ 0.5, wm_high ≈ 0.8; start with a=b=0.5; tune target batch size b_target from the ratio c(e)/w(v).

XII. Interface Bindings and Invariants (Aligned with I70-3)

1. API sketch
   • chan_open(name:str, cap:int, mode:str="mpmc") -> ChanRef (validate cap>0).
   • chan_put(ch, msg|batch, key?, timeout?) -> bool (timeouts increment TS.timeout.count).
   • chan_get(ch, max_batch:int=1, timeout?) -> list (empty-timeout increments TS.timeout.count).
   • set_backpressure(ch, strategy:dict) -> None (strategy includes wm_low, wm_high, actions[]).
2. Invariants
   q_len <= cap; bp ∈ [0,1]; ACK is serialized or idempotent; trace_link(span,eid) is complete.

XIII. Observability and SLIs

1. TS. mappings*
   • TS.chan.cap, TS.chan.q_len, TS.backpressure.level, TS.drop.count, TS.retry.count, TS.timeout.count.
   • Latency: TS.sli.queue_time_ms, TS.sli.latency_ms, TS.sli.p99_ms; throughput: TS.sli.qps; errors: TS.sli.err_rate.
2. Dimensional compliance
   Verify check_dim( L - lambda * W ) = 0; window is SLA_window.

XIV. Contracts and Test Matrix

1. Must-pass cases
   • Steady state with rho < 1; watermark wm_high triggers rate limit; head_drop/tail_drop/random_drop correctness.
   • Effective dedup under at_least_once; ack_timeout retries preserve ordering constraints; Delta_t_dedup takes effect.
   • Aging and expiry: ttl expiry drops are audited.
2. Assertion structure (example)
   assert_thread_contract(G, tests=[ {"type":"queue_stability","chan":"ingress","rho_lt":1.0}, {"type":"slo","p99_ms_le":SLO}, {"type":"drop_policy","mode":"head_drop"} ]).

XV. Operating Flow Mx-2 (Closed-Loop Backpressure)

• chan_open, configure cap, wm_low/wm_high, strategy.
• Periodically sample q_len, W_q, bp, compute bp = f(q_len, cap, W_q).
• If bp >= wm_high: execute actions_high = { limit rps, enlarge batch, prefetch↓ }.
• If bp <= wm_low: execute actions_low = { relax rps, shrink batch, prefetch↑ }.
• Monitor TS.drop.count / err_rate / P99; if needed, trigger shed_by_key or scale up K_thr.
• Run assert_thread_contract; persist Trace and TS.*.

XVI. Exit Criteria and Deliverables

• Pass P73-, S73-, and the full Mx-2 loop; TS.* metrics complete; overflow and ACK audits are traceable.
• Deliver: channel spec, capacity and watermark tables, drop and retry strategy cards, dedup configuration, and regression baselines.