The buffer absorbed the signal backpressure needed

# Self-authored for the EML case corpus (no external origin). A queue has both
# a buffer to absorb bursts and backpressure to slow the producer. When each
# one engages is computed below.
#
# Both are correct. The buffer was sized from a measured burst profile and it
# absorbs the twice-daily spike that used to drop messages. Backpressure was
# added because an unbounded producer once filled the disk. Each was a good
# change and neither team was guessing.
#
# Backpressure engages on observed lag. The buffer's entire purpose is to keep
# lag from being observed during a burst. So the buffer hides the signal for
# exactly as long as it has capacity, and backpressure engages only after the
# buffer is exhausted - which is the moment when slowing down helps least,
# because there is already a full buffer of work to drain.
#
# Depth and latency are computed per minute across one sustained overload.

# [minute, produced per second, consumed per second, buffer capacity]
[["t+0", 800, 1000, 60000], ["t+1", 1400, 1000, 60000], ["t+2", 1400, 1000, 60000], ["t+3", 1400, 1000, 60000], ["t+4", 1400, 1000, 60000], ["t+5", 1400, 1000, 60000], ["t+6", 1400, 1000, 60000]] => window

len(window) => n
55 => lag_threshold

"minute   produced/s   consumed/s   net/s   buffer depth   lag seconds   backpressure" ^0
0 => depth
0 => engaged_at
"" => engaged_min
for w in window:
    w[1] - w[2] => net
    depth + net * 60 => depth
    if depth < 0:
        0 => depth
    if depth > w[3]:
        w[3] => depth
    int(depth / w[2]) => lag
    "" => bp
    if lag > lag_threshold:
        "ON" => bp
        if engaged_min == "":
            w[0] => engaged_min
    else:
        "off" => bp
    "  " + w[0] + "     " + str(w[1]) + "         " + str(w[2]) + "         " + str(net) + "     " + str(depth) + "         " + str(lag) + "            " + bp ^0
"" ^0

"buffer capacity      : " + str(window[0][3]) + " messages" ^0
"backpressure engages : lag above " + str(lag_threshold) + " seconds" ^0
if not (engaged_min == ""):
    "first engaged at     : " + engaged_min ^0
else:
    "backpressure never engaged inside this window" ^0
"" ^0

# ---- how long the buffer hides the overload ----

window[1][1] - window[1][2] => surplus
int(window[0][3] / (surplus * 60)) => minutes_hidden
"how long the buffer conceals a sustained " + str(surplus) + "/s surplus" ^0
"  buffer capacity / surplus per minute : " + str(window[0][3]) + " / " + str(surplus * 60) ^0
"  minutes before the buffer is full    : " + str(minutes_hidden) ^0
"  during those minutes the observed lag rises from 0, so the signal is not" ^0
"  absent - it is present and below the threshold that acts on it" ^0
"" ^0

# ---- what backpressure can still do when it engages ----

"the state at the moment backpressure engages" ^0
"  buffer depth   : " + str(window[0][3]) + " messages" ^0
"  drain rate     : " + str(window[0][2]) + "/s" ^0
"  time to drain, with the producer stopped completely : " + str(int(window[0][3] / window[0][2])) + " seconds" ^0
"  the oldest message in the buffer is already that old" ^0
"  slowing the producer now removes future arrivals and does nothing about" ^0
"  the queue that is already there" ^0
"" ^0

# ---- the same overload with a smaller buffer ----

6000 => small
int(small / (surplus * 60)) => small_hidden
"the same producer against a " + str(small) + "-message buffer" ^0
"  minutes before full : " + str(small_hidden) ^0
"  drain time at engagement : " + str(int(small / window[0][2])) + " seconds" ^0
"  backpressure engages " + str(minutes_hidden - small_hidden) + " minutes earlier and with " + str(int(window[0][3] / window[0][2]) - int(small / window[0][2])) + " seconds" ^0
"  less accumulated work to clear" ^0
"  the smaller buffer drops more during the twice-daily burst, which is the" ^0
"  thing the large one was bought for" ^0
"" ^0

# ---- the two settings are one decision ----

"buffer size against how late backpressure can act" ^0
[6000, 20000, 60000, 120000] => sizes
for s in sizes:
    "  " + str(s) + " messages : hidden for " + str(int(s / (surplus * 60))) + " min, " + str(int(s / window[0][2])) + " s of work at engagement" ^0
"  the buffer is sized by the burst profile and the engagement delay is a" ^0
"  consequence nobody chose" ^0
"" ^0

# ---- what would engage on the right quantity ----

"a signal that a buffer cannot absorb" ^0
"  observed lag       : absorbed, by design, that is what the buffer is" ^0
"  buffer depth       : rises immediately, already recorded" ^0
"  arrival minus service rate : rises immediately, already recorded" ^0
"  controls reading either of the second two : 0" ^0
"  the rate difference is visible in the first minute of the overload and" ^0
"  the control waits for a quantity the buffer is built to suppress" ^0
"" ^0

# ---- the control: a burst the buffer was actually sized for ----
#
# Where the surplus is temporary, the buffer absorbs it, lag never rises and
# backpressure correctly never engages.

[["burst", 1800, 1000, 60000], ["after-1", 400, 1000, 60000], ["after-2", 400, 1000, 60000]] => burst
"control - a 60-second burst rather than a sustained overload" ^0
0 => bdepth
0 => bpeak
0 => bmax_lag
for b in burst:
    b[1] - b[2] => net
    bdepth + net * 60 => bdepth
    if bdepth < 0:
        0 => bdepth
    if bdepth > bpeak:
        bdepth => bpeak
    int(bdepth / b[2]) => blag
    if blag > bmax_lag:
        blag => bmax_lag
    "  " + b[0] + " : net " + str(net) + "/s, depth " + str(bdepth) + ", lag " + str(blag) + "s" ^0
"  peak depth " + str(bpeak) + " against capacity " + str(burst[0][3]) + ", which is " + str(int(bpeak * 100 / burst[0][3])) + "%" ^0
"  peak lag " + str(bmax_lag) + "s against the " + str(lag_threshold) + "s threshold" ^0
if bdepth == 0:
    "  and the queue returns to empty on its own" ^0
"  backpressure never engaged, and it was right not to" ^0
"  the same buffer, the same rule, and the difference is whether the" ^0
"  surplus ends on its own" ^0
"" ^0

"The buffer absorbs the burst it was measured for and backpressure stopped a" ^0
"disk from filling. Backpressure waits on lag and the buffer's job is to" ^0
"absorb lag, so the second control cannot act until the first has run out." ^0

window = [["t+0", 800, 1000, 60000], ["t+1", 1400, 1000, 60000], ["t+2", 1400, 1000, 60000], ["t+3", 1400, 1000, 60000], ["t+4", 1400, 1000, 60000], ["t+5", 1400, 1000, 60000], ["t+6", 1400, 1000, 60000]]
n = len(window)
lag_threshold = 55
print("minute   produced/s   consumed/s   net/s   buffer depth   lag seconds   backpressure")
depth = 0
engaged_at = 0
engaged_min = ""
for w in window:
    net = w[1] - w[2]
    depth = depth + net * 60
    if depth < 0:
        depth = 0
    if depth > w[3]:
        depth = w[3]
    lag = int(depth / w[2])
    bp = ""
    if lag > lag_threshold:
        bp = "ON"
        if engaged_min == "":
            engaged_min = w[0]
    else:
        bp = "off"
    print("  " + w[0] + "     " + str(w[1]) + "         " + str(w[2]) + "         " + str(net) + "     " + str(depth) + "         " + str(lag) + "            " + bp)
print("")
print("buffer capacity      : " + str(window[0][3]) + " messages")
print("backpressure engages : lag above " + str(lag_threshold) + " seconds")
if not engaged_min == "":
    print("first engaged at     : " + engaged_min)
else:
    print("backpressure never engaged inside this window")
print("")
surplus = window[1][1] - window[1][2]
minutes_hidden = int(window[0][3] / (surplus * 60))
print("how long the buffer conceals a sustained " + str(surplus) + "/s surplus")
print("  buffer capacity / surplus per minute : " + str(window[0][3]) + " / " + str(surplus * 60))
print("  minutes before the buffer is full    : " + str(minutes_hidden))
print("  during those minutes the observed lag rises from 0, so the signal is not")
print("  absent - it is present and below the threshold that acts on it")
print("")
print("the state at the moment backpressure engages")
print("  buffer depth   : " + str(window[0][3]) + " messages")
print("  drain rate     : " + str(window[0][2]) + "/s")
print("  time to drain, with the producer stopped completely : " + str(int(window[0][3] / window[0][2])) + " seconds")
print("  the oldest message in the buffer is already that old")
print("  slowing the producer now removes future arrivals and does nothing about")
print("  the queue that is already there")
print("")
small = 6000
small_hidden = int(small / (surplus * 60))
print("the same producer against a " + str(small) + "-message buffer")
print("  minutes before full : " + str(small_hidden))
print("  drain time at engagement : " + str(int(small / window[0][2])) + " seconds")
print("  backpressure engages " + str(minutes_hidden - small_hidden) + " minutes earlier and with " + str(int(window[0][3] / window[0][2]) - int(small / window[0][2])) + " seconds")
print("  less accumulated work to clear")
print("  the smaller buffer drops more during the twice-daily burst, which is the")
print("  thing the large one was bought for")
print("")
print("buffer size against how late backpressure can act")
sizes = [6000, 20000, 60000, 120000]
for s in sizes:
    print("  " + str(s) + " messages : hidden for " + str(int(s / (surplus * 60))) + " min, " + str(int(s / window[0][2])) + " s of work at engagement")
print("  the buffer is sized by the burst profile and the engagement delay is a")
print("  consequence nobody chose")
print("")
print("a signal that a buffer cannot absorb")
print("  observed lag       : absorbed, by design, that is what the buffer is")
print("  buffer depth       : rises immediately, already recorded")
print("  arrival minus service rate : rises immediately, already recorded")
print("  controls reading either of the second two : 0")
print("  the rate difference is visible in the first minute of the overload and")
print("  the control waits for a quantity the buffer is built to suppress")
print("")
burst = [["burst", 1800, 1000, 60000], ["after-1", 400, 1000, 60000], ["after-2", 400, 1000, 60000]]
print("control - a 60-second burst rather than a sustained overload")
bdepth = 0
bpeak = 0
bmax_lag = 0
for b in burst:
    net = b[1] - b[2]
    bdepth = bdepth + net * 60
    if bdepth < 0:
        bdepth = 0
    if bdepth > bpeak:
        bpeak = bdepth
    blag = int(bdepth / b[2])
    if blag > bmax_lag:
        bmax_lag = blag
    print("  " + b[0] + " : net " + str(net) + "/s, depth " + str(bdepth) + ", lag " + str(blag) + "s")
print("  peak depth " + str(bpeak) + " against capacity " + str(burst[0][3]) + ", which is " + str(int(bpeak * 100 / burst[0][3])) + "%")
print("  peak lag " + str(bmax_lag) + "s against the " + str(lag_threshold) + "s threshold")
if bdepth == 0:
    print("  and the queue returns to empty on its own")
print("  backpressure never engaged, and it was right not to")
print("  the same buffer, the same rule, and the difference is whether the")
print("  surplus ends on its own")
print("")
print("The buffer absorbs the burst it was measured for and backpressure stopped a")
print("disk from filling. Backpressure waits on lag and the buffer's job is to")
print("absorb lag, so the second control cannot act until the first has run out.")

minute   produced/s   consumed/s   net/s   buffer depth   lag seconds   backpressure
  t+0     800         1000         -200     0         0            off
  t+1     1400         1000         400     24000         24            off
  t+2     1400         1000         400     48000         48            off
  t+3     1400         1000         400     60000         60            ON
  t+4     1400         1000         400     60000         60            ON
  t+5     1400         1000         400     60000         60            ON
  t+6     1400         1000         400     60000         60            ON

buffer capacity      : 60000 messages
backpressure engages : lag above 55 seconds
first engaged at     : t+3

how long the buffer conceals a sustained 400/s surplus
  buffer capacity / surplus per minute : 60000 / 24000
  minutes before the buffer is full    : 2
  during those minutes the observed lag rises from 0, so the signal is not
  absent - it is present and below the threshold that acts on it

the state at the moment backpressure engages
  buffer depth   : 60000 messages
  drain rate     : 1000/s
  time to drain, with the producer stopped completely : 60 seconds
  the oldest message in the buffer is already that old
  slowing the producer now removes future arrivals and does nothing about
  the queue that is already there

the same producer against a 6000-message buffer
  minutes before full : 0
  drain time at engagement : 6 seconds
  backpressure engages 2 minutes earlier and with 54 seconds
  less accumulated work to clear
  the smaller buffer drops more during the twice-daily burst, which is the
  thing the large one was bought for

buffer size against how late backpressure can act
  6000 messages : hidden for 0 min, 6 s of work at engagement
  20000 messages : hidden for 0 min, 20 s of work at engagement
  60000 messages : hidden for 2 min, 60 s of work at engagement
  120000 messages : hidden for 5 min, 120 s of work at engagement
  the buffer is sized by the burst profile and the engagement delay is a
  consequence nobody chose

a signal that a buffer cannot absorb
  observed lag       : absorbed, by design, that is what the buffer is
  buffer depth       : rises immediately, already recorded
  arrival minus service rate : rises immediately, already recorded
  controls reading either of the second two : 0
  the rate difference is visible in the first minute of the overload and
  the control waits for a quantity the buffer is built to suppress

control - a 60-second burst rather than a sustained overload
  burst : net 800/s, depth 48000, lag 48s
  after-1 : net -600/s, depth 12000, lag 12s
  after-2 : net -600/s, depth 0, lag 0s
  peak depth 48000 against capacity 60000, which is 80%
  peak lag 48s against the 55s threshold
  and the queue returns to empty on its own
  backpressure never engaged, and it was right not to
  the same buffer, the same rule, and the difference is whether the
  surplus ends on its own

The buffer absorbs the burst it was measured for and backpressure stopped a
disk from filling. Backpressure waits on lag and the buffer's job is to
absorb lag, so the second control cannot act until the first has run out.

Trace event types