Project P117
Message queue
A broker that sends a message again when its acknowledgement is lost, four handlers over the same deliveries - add, replay, check and a window of keys - and every choice of lost acks counted, with a key reused for another request refused.
2 modules · 2 recorded sessionstext-menu UI in the terminalupdated 2026-10-16
Every screen below was recorded under CPython. When this page was built, the EML interpreter replayed each session from the same input and printed the same bytes.
About
A broker delivers every message at least once. When a consumer's ack is lost, the broker sends the message again, and the work is done twice unless the handler notices. Each message carries an idempotency key.
main.eml- the menu, the messages, the deliveries with four handlers side by side, every choice of lost acks, and the settingsbroker.eml- the messages, the right result, the broker's order of deliveries, the four handlers, and the count over every choice
How each part works:
- A message is a key, an account and an amount; every account starts at 0. A lost ack sends the message again a set number of deliveries later, or sooner when nothing else is waiting. The ack can be lost again.
- The right result applies each key's first message once. A later message with the same key and the same request is a retry, answered done; with another request, it is refused.
- The four handlers, as in the corpus cases
at-least-once-deliveryandidempotency-key-reuse: - add applies every delivery, so a duplicate is applied again; - replay keeps every key and answers a key it has seen with the first answer: the balances are right, but a reused key is told done; - check keeps every key with its request and refuses another request under it: right on every run, and it keeps every key; - a window keeps only the last few keys, so it is right only while every duplicate comes back before its key has left. When a key has left and come back with another request, the window can even refuse a right one. - Every choice: each message loses its first ack or not, 128 runs for the seven messages. With a message sent again 3 deliveries later, the first window right on every run keeps 3 keys; 6 deliveries later, 4. For every delay from 1 to 9 it is 2, 3, 3, 4, 4, 4, 5, 5 and 5 keys.
What is checked: menu choices 1 to 7; lost acks as message numbers, at most 3 for one message, or none; a key of up to 8 letters and digits, an account of up to 10 lowercase letters, an amount from -999 to 999 but not 0, and at most 8 messages; a delay and a window from 1 to 9. An empty answer cancels.
Sessions: sessions/basic.in gives: - the messages, with one key reused for another request and one retry; - deliveries losing the acks of messages 2 and 5, of message 1 (the window of 2 is too small), and of none; - every choice at a delay of 3, and of 6; - a window of 4, with message 1's ack lost once and message 2's twice; - message 8, a late retry of message 1, after which only a window of all 5 keys is right on every run.
sessions/bad-input.in gives: - menu choices 0 and x; - lost acks: a cancel, 0, 8, x, 2,5 and 1 four times; - new messages: a cancel, the keys k 1, toolongkey and k-1, a cancel, the accounts Bob and averyverylongname, a cancel, the amounts 0, 1000, +-3 and x, then dave -15, and a ninth message; - the delay: a cancel, 0, 10 and x; the window: a cancel and 0.
Built on the verified corpus cases at-least-once-delivery (duplicates are the handler's problem, and a bounded window is an assumption about lateness) and idempotency-key-reuse (a reused key with another request is refused only when the request is kept).
Recorded sessions
What the screen shows while someone uses the program. Each typed line appears after its prompt, the way a terminal shows it.
bad-input
interpreter: byte-equal== Message queue ==
A broker delivers every message at least once: when an ack is lost, it sends the message again. Each message carries an idempotency key.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 0
Pick a number from 1 to 7.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> x
Pick a number from 1 to 7.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 2
Acks lost: message numbers from 1 to 7 such as 2 5 (a number twice loses two), or none.
acks lost>
Cancelled.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 2
Acks lost: message numbers from 1 to 7 such as 2 5 (a number twice loses two), or none.
acks lost> 0
Type message numbers from 1 to 7, or none.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 2
Acks lost: message numbers from 1 to 7 such as 2 5 (a number twice loses two), or none.
acks lost> 8
Type message numbers from 1 to 7, or none.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 2
Acks lost: message numbers from 1 to 7 such as 2 5 (a number twice loses two), or none.
acks lost> x
Type message numbers from 1 to 7, or none.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 2
Acks lost: message numbers from 1 to 7 such as 2 5 (a number twice loses two), or none.
acks lost> 2,5
Type message numbers from 1 to 7, or none.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 2
Acks lost: message numbers from 1 to 7 such as 2 5 (a number twice loses two), or none.
acks lost> 1 1 1 1
At most 3 lost acks for one message.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 4
key (up to 8 letters and digits)>
Cancelled.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 4
key (up to 8 letters and digits)> k 1
Type a key of up to 8 letters and digits.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 4
key (up to 8 letters and digits)> toolongkey
Type a key of up to 8 letters and digits.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 4
key (up to 8 letters and digits)> k-1
Type a key of up to 8 letters and digits.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 4
key (up to 8 letters and digits)> k9
account (up to 10 lowercase letters)>
Cancelled.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 4
key (up to 8 letters and digits)> k9
account (up to 10 lowercase letters)> Bob
Type an account of up to 10 lowercase letters.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 4
key (up to 8 letters and digits)> k9
account (up to 10 lowercase letters)> averyverylongname
Type an account of up to 10 lowercase letters.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 4
key (up to 8 letters and digits)> k9
account (up to 10 lowercase letters)> dave
amount (a whole number from -999 to 999, not 0)>
Cancelled.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 4
key (up to 8 letters and digits)> k9
account (up to 10 lowercase letters)> dave
amount (a whole number from -999 to 999, not 0)> 0
Type a whole number from -999 to 999, not 0.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 4
key (up to 8 letters and digits)> k9
account (up to 10 lowercase letters)> dave
amount (a whole number from -999 to 999, not 0)> 1000
Type a whole number from -999 to 999, not 0.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 4
key (up to 8 letters and digits)> k9
account (up to 10 lowercase letters)> dave
amount (a whole number from -999 to 999, not 0)> +-3
Type a whole number from -999 to 999, not 0.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 4
key (up to 8 letters and digits)> k9
account (up to 10 lowercase letters)> dave
amount (a whole number from -999 to 999, not 0)> x
Type a whole number from -999 to 999, not 0.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 4
key (up to 8 letters and digits)> k9
account (up to 10 lowercase letters)> dave
amount (a whole number from -999 to 999, not 0)> -15
Added message 8: k9 dave -15.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 4
8 messages is the most.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 5
sent again after (1 to 9 deliveries)>
Cancelled.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 5
sent again after (1 to 9 deliveries)> 0
Type a number from 1 to 9.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 5
sent again after (1 to 9 deliveries)> 10
Type a number from 1 to 9.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 5
sent again after (1 to 9 deliveries)> x
Type a number from 1 to 9.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 6
window (1 to 9 keys)>
Cancelled.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 6
window (1 to 9 keys)> 0
Type a number from 1 to 9.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 7
Bye.
What was typed (69 lines)
0
x
2
2
0
2
8
2
x
2
2,5
2
1 1 1 1
4
4
k 1
4
toolongkey
4
k-1
4
k9
4
k9
Bob
4
k9
averyverylongname
4
k9
dave
4
k9
dave
0
4
k9
dave
1000
4
k9
dave
+-3
4
k9
dave
x
4
k9
dave
-15
4
5
5
0
5
10
5
x
6
6
0
7
basic
interpreter: byte-equal== Message queue ==
A broker delivers every message at least once: when an ack is lost, it sends the message again. Each message carries an idempotency key.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 1
Messages (a message whose ack is lost is sent again 3 deliveries later; the window keeps 2 keys):
1 k1 alice +30
2 k2 bob +20
3 k3 alice -10
4 k2 bob +25 the key of message 2, for another request
5 k4 carol +50
6 k4 carol +50 a retry of message 5
7 k5 bob -5
Right: alice 20, bob 15, carol 50; message 4 refused.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 2
Acks lost: message numbers from 1 to 7 such as 2 5 (a number twice loses two), or none.
acks lost> 2 5
Deliveries:
1 message 1 k1 alice +30
2 message 2 k2 bob +20 ack lost
3 message 3 k3 alice -10
4 message 4 k2 bob +25
5 message 2 k2 bob +20 sent again
6 message 5 k4 carol +50 ack lost
7 message 6 k4 carol +50
8 message 7 k5 bob -5
9 message 5 k4 carol +50 sent again
Right: alice 20, bob 15, carol 50; message 4 refused.
add alice 20, bob 60, carol 150: 3 applied again; message 4 applied under message 2's key.
replay alice 20, bob 15, carol 50: message 4 told done with message 2's answer.
check alice 20, bob 15, carol 50: right, keeping 5 keys with their requests.
window 2 alice 20, bob 15, carol 50: right, keeping 2 keys with their requests.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 2
Acks lost: message numbers from 1 to 7 such as 2 5 (a number twice loses two), or none.
acks lost> 1
Deliveries:
1 message 1 k1 alice +30 ack lost
2 message 2 k2 bob +20
3 message 3 k3 alice -10
4 message 1 k1 alice +30 sent again
5 message 4 k2 bob +25
6 message 5 k4 carol +50
7 message 6 k4 carol +50
8 message 7 k5 bob -5
Right: alice 20, bob 15, carol 50; message 4 refused.
add alice 50, bob 40, carol 100: 2 applied again; message 4 applied under message 2's key.
replay alice 20, bob 15, carol 50: message 4 told done with message 2's answer.
check alice 20, bob 15, carol 50: right, keeping 5 keys with their requests.
window 2 alice 50, bob 40, carol 50: 1 applied again; message 4 applied under message 2's key.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 2
Acks lost: message numbers from 1 to 7 such as 2 5 (a number twice loses two), or none.
acks lost> none
Deliveries:
1 message 1 k1 alice +30
2 message 2 k2 bob +20
3 message 3 k3 alice -10
4 message 4 k2 bob +25
5 message 5 k4 carol +50
6 message 6 k4 carol +50
7 message 7 k5 bob -5
Right: alice 20, bob 15, carol 50; message 4 refused.
add alice 20, bob 40, carol 100: 1 applied again; message 4 applied under message 2's key.
replay alice 20, bob 15, carol 50: message 4 told done with message 2's answer.
check alice 20, bob 15, carol 50: right, keeping 5 keys with their requests.
window 2 alice 20, bob 15, carol 50: right, keeping 2 keys with their requests.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 3
Every choice of which messages lose their first ack: 128 runs, each lost one sent again 3 deliveries later, or sooner at the end.
handler right applied twice reuse mishandled keys kept
add 0 128 128 0
replay 0 0 128 5
check 128 0 0 5
window 1 0 124 120 1
window 2 40 74 64 2
window 3 128 0 0 3
applied twice: some request was applied more than once.
reuse mishandled: a key reused for another request was not refused, or a message was wrongly refused.
The first window right on every run keeps 3 keys; check keeps 5. How small a window can be depends on how late a message comes back.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 5
sent again after (1 to 9 deliveries)> 6
A message whose ack is lost is now sent again 6 deliveries later.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 3
Every choice of which messages lose their first ack: 128 runs, each lost one sent again 6 deliveries later, or sooner at the end.
handler right applied twice reuse mishandled keys kept
add 0 128 128 0
replay 0 0 128 5
check 128 0 0 5
window 1 0 126 128 1
window 2 8 118 32 2
window 3 16 96 32 3
window 4 128 0 0 4
applied twice: some request was applied more than once.
reuse mishandled: a key reused for another request was not refused, or a message was wrongly refused.
The first window right on every run keeps 4 keys; check keeps 5. How small a window can be depends on how late a message comes back.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 6
window (1 to 9 keys)> 4
The window now keeps 4 keys.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 2
Acks lost: message numbers from 1 to 7 such as 2 5 (a number twice loses two), or none.
acks lost> 1 2 2
Deliveries:
1 message 1 k1 alice +30 ack lost
2 message 2 k2 bob +20 ack lost
3 message 3 k3 alice -10
4 message 4 k2 bob +25
5 message 5 k4 carol +50
6 message 6 k4 carol +50
7 message 1 k1 alice +30 sent again
8 message 2 k2 bob +20 sent again, ack lost
9 message 7 k5 bob -5
10 message 2 k2 bob +20 sent again
Right: alice 20, bob 15, carol 50; message 4 refused.
add alice 50, bob 80, carol 100: 4 applied again; message 4 applied under message 2's key.
replay alice 20, bob 15, carol 50: message 4 told done with message 2's answer.
check alice 20, bob 15, carol 50: right, keeping 5 keys with their requests.
window 4 alice 20, bob 15, carol 50: right, keeping 4 keys with their requests.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 4
key (up to 8 letters and digits)> k1
account (up to 10 lowercase letters)> alice
amount (a whole number from -999 to 999, not 0)> 30
Added message 8: k1 alice +30, a retry of message 1.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 1
Messages (a message whose ack is lost is sent again 6 deliveries later; the window keeps 4 keys):
1 k1 alice +30
2 k2 bob +20
3 k3 alice -10
4 k2 bob +25 the key of message 2, for another request
5 k4 carol +50
6 k4 carol +50 a retry of message 5
7 k5 bob -5
8 k1 alice +30 a retry of message 1
Right: alice 20, bob 15, carol 50; message 4 refused.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 3
Every choice of which messages lose their first ack: 256 runs, each lost one sent again 6 deliveries later, or sooner at the end.
handler right applied twice reuse mishandled keys kept
add 0 256 256 0
replay 0 0 256 5
check 256 0 0 5
window 1 0 256 256 1
window 2 0 256 64 2
window 3 0 256 64 3
window 4 0 256 64 4
window 5 256 0 0 5
applied twice: some request was applied more than once.
reuse mishandled: a key reused for another request was not refused, or a message was wrongly refused.
The first window right on every run keeps 5 keys; check keeps 5. How small a window can be depends on how late a message comes back.
1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit
choice> 7
Bye.
What was typed (22 lines)
1
2
2 5
2
1
2
none
3
5
6
3
6
4
2
1 2 2
4
k1
alice
30
1
3
7
Modules
The program as written, entry module first. Each module transpiles to its own Python file, which is what eml project run executes.
main.eml(entry)
eml# P117 message queue: a broker that sends a message again when its
# acknowledgement is lost, four handlers side by side over the same
# deliveries, and every choice of lost acknowledgements counted.
import broker
def plain(s):
0 => i
len(s) => j
while i < j and s[i] == " ":
i + 1 => i
while j > i and s[j - 1] == " ":
j - 1 => j
return s[i:j]
def pad(s, n):
s => out
while len(out) < n:
out + " " => out
return out
def lpad(s, n):
s => out
while len(out) < n:
" " + out => out
return out
def is_digits(s):
if len(s) == 0:
return False
for c in s:
if c < "0" or c > "9":
return False
return True
def numbers_text(ns):
"" => out
for i in [0:len(ns) - 1]:
if i > 0 and i == len(ns) - 1:
out + " and " => out
elif i > 0:
out + ", " => out
out + str(ns[i]) => out
return out
def balances_text(msgs, bal):
"" => out
for a in broker.accounts(msgs):
if out != "":
out + ", " => out
out + a + " " + str(bal[a]) => out
return out
def note(msgs, i):
broker.first_with_key(msgs, i) => f
if f == i:
return ""
if broker.same_request(msgs[f], msgs[i]):
return "a retry of message " + str(f + 1)
return "the key of message " + str(f + 1) + ", for another request"
def widths(msgs):
0 => kw
0 => bw
for m in msgs:
if len(m[0]) > kw:
len(m[0]) => kw
if len(broker.body_text(m)) > bw:
len(broker.body_text(m)) => bw
return [kw, bw]
def right_text(msgs):
broker.expected(msgs) => want
[] => refused
for i in [0:len(msgs) - 1]:
if want[1][i] == "refused":
refused + [i + 1] => refused
"Right: " + balances_text(msgs, want[0]) => out
if len(refused) == 1:
out + "; message " + numbers_text(refused) + " refused" => out
elif len(refused) > 1:
out + "; messages " + numbers_text(refused) + " refused" => out
return out + "."
def show(msgs, delay, k):
("Messages (a message whose ack is lost is sent again " + str(delay) + " deliveries later; the window keeps " + str(k) + " keys):") ^0
widths(msgs) => w
for i in [0:len(msgs) - 1]:
msgs[i] => m
" " + lpad(str(i + 1), 2) + " " + pad(m[0], w[0]) + " " => line
note(msgs, i) => n
if n == "":
(line + broker.body_text(m)) ^0
else:
(line + pad(broker.body_text(m), w[1]) + " " + n) ^0
right_text(msgs) ^0
def handler_line(msgs, name, kind, k, seq):
broker.handle(kind, k, msgs, seq) => r
broker.problems(msgs, r) => p
[] => parts
if p[0] > 0:
parts + [str(p[0]) + " applied again"] => parts
for x in p[1]:
"message " + str(x[0] + 1) => who
if x[1] == "applied":
parts + [who + " applied under message " + str(x[2] + 1) + "'s key"] => parts
elif x[1] == "told":
parts + [who + " told done with message " + str(x[2] + 1) + "'s answer"] => parts
else:
parts + [who + " wrongly refused"] => parts
if len(parts) == 0 and not broker.balances_right(msgs, r):
parts + ["balances wrong"] => parts
"" => verdict
if len(parts) == 0:
"right" => verdict
if kind == "replay":
verdict + ", keeping " + str(r[4]) + " keys" => verdict
elif kind != "add":
verdict + ", keeping " + str(r[4]) + " keys with their requests" => verdict
else:
for q in parts:
if verdict != "":
verdict + "; " => verdict
verdict + q => verdict
(" " + pad(name, 10) + balances_text(msgs, r[0]) + ": " + verdict + ".") ^0
def deliver(msgs, delay, k):
("Acks lost: message numbers from 1 to " + str(len(msgs)) + " such as 2 5 (a number twice loses two), or none.") ^0
plain(input("acks lost> ")) => t
if t == "":
"Cancelled." ^0
return
[] => lost
for m in msgs:
lost + [0] => lost
if t != "none":
"" => w
for c in t + " ":
if c == " ":
if w != "":
if not is_digits(w) or len(w) > 2 or int(w) < 1 or int(w) > len(msgs):
("Type message numbers from 1 to " + str(len(msgs)) + ", or none.") ^0
return
lost[int(w) - 1] + 1 => lost[int(w) - 1]
if lost[int(w) - 1] > 3:
"At most 3 lost acks for one message." ^0
return
"" => w
else:
w + c => w
broker.deliveries(msgs, lost, delay) => seq
widths(msgs) => wd
"Deliveries:" ^0
for d in [0:len(seq) - 1]:
seq[d] => e
msgs[e[0]] => m
"" => tag
if e[2]:
"sent again" => tag
if e[1]:
if tag != "":
tag + ", " => tag
tag + "ack lost" => tag
" " + lpad(str(d + 1), 2) + " message " + str(e[0] + 1) + " " + pad(m[0], wd[0]) + " " => line
if tag == "":
(line + broker.body_text(m)) ^0
else:
(line + pad(broker.body_text(m), wd[1]) + " " + tag) ^0
right_text(msgs) ^0
handler_line(msgs, "add", "add", k, seq)
handler_line(msgs, "replay", "replay", k, seq)
handler_line(msgs, "check", "check", k, seq)
handler_line(msgs, "window " + str(k), "window", k, seq)
def every(msgs, delay):
broker.census(msgs, delay) => c
("Every choice of which messages lose their first ack: " + str(c[0]) + " runs, each lost one sent again " + str(delay) + " deliveries later, or sooner at the end.") ^0
" handler right applied twice reuse mishandled keys kept" ^0
0 => kept
for row in c[1]:
(" " + pad(row[0], 10) + lpad(str(row[1]), 8) + lpad(str(row[2]), 15) + lpad(str(row[3]), 18) + lpad(str(row[4]), 11)) ^0
if row[0] == "check":
row[4] => kept
" applied twice: some request was applied more than once." ^0
" reuse mishandled: a key reused for another request was not refused, or a message was wrongly refused." ^0
c[1][len(c[1]) - 1] => last
("The first window right on every run keeps " + str(last[4]) + " keys; check keeps " + str(kept) + ". How small a window can be depends on how late a message comes back.") ^0
def add_message(msgs):
if len(msgs) >= 8:
"8 messages is the most." ^0
return msgs
plain(input("key (up to 8 letters and digits)> ")) => key
if key == "":
"Cancelled." ^0
return msgs
if len(key) > 8:
"Type a key of up to 8 letters and digits." ^0
return msgs
for c in key:
if not ((c >= "a" and c <= "z") or (c >= "A" and c <= "Z") or (c >= "0" and c <= "9")):
"Type a key of up to 8 letters and digits." ^0
return msgs
plain(input("account (up to 10 lowercase letters)> ")) => acct
if acct == "":
"Cancelled." ^0
return msgs
if len(acct) > 10:
"Type an account of up to 10 lowercase letters." ^0
return msgs
for c in acct:
if c < "a" or c > "z":
"Type an account of up to 10 lowercase letters." ^0
return msgs
plain(input("amount (a whole number from -999 to 999, not 0)> ")) => t
if t == "":
"Cancelled." ^0
return msgs
1 => sign
t => digits
if t[0] == "-" or t[0] == "+":
if t[0] == "-":
0 - 1 => sign
t[1:len(t)] => digits
if not is_digits(digits) or len(digits) > 3 or int(digits) == 0:
"Type a whole number from -999 to 999, not 0." ^0
return msgs
msgs + [[key, acct, sign * int(digits)]] => out
note(out, len(out) - 1) => n
if n == "":
("Added message " + str(len(out)) + ": " + key + " " + broker.body_text(out[len(out) - 1]) + ".") ^0
else:
("Added message " + str(len(out)) + ": " + key + " " + broker.body_text(out[len(out) - 1]) + ", " + n + ".") ^0
return out
def one_to_nine(prompt):
plain(input(prompt)) => t
if t == "":
"Cancelled." ^0
return 0
if len(t) != 1 or t < "1" or t > "9":
"Type a number from 1 to 9." ^0
return 0
return int(t)
broker.default_messages() => msgs
3 => delay
2 => k
"== Message queue ==" ^0
"A broker delivers every message at least once: when an ack is lost, it sends the message again. Each message carries an idempotency key." ^0
True => running
while running:
"" ^0
"1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit" ^0
plain(input("choice> ")) => choice
if choice == "1":
show(msgs, delay, k)
elif choice == "2":
deliver(msgs, delay, k)
elif choice == "3":
every(msgs, delay)
elif choice == "4":
add_message(msgs) => msgs
elif choice == "5":
one_to_nine("sent again after (1 to 9 deliveries)> ") => n
if n > 0:
n => delay
("A message whose ack is lost is now sent again " + str(delay) + " deliveries later.") ^0
elif choice == "6":
one_to_nine("window (1 to 9 keys)> ") => n
if n > 0:
n => k
("The window now keeps " + str(k) + " keys.") ^0
elif choice == "7":
False => running
else:
"Pick a number from 1 to 7." ^0
"Bye." ^0
Python projection (main.py)
import broker
def plain(s):
i = 0
j = len(s)
while i < j and s[i] == " ":
i = i + 1
while j > i and s[j - 1] == " ":
j = j - 1
return s[i:j]
def pad(s, n):
out = s
while len(out) < n:
out = out + " "
return out
def lpad(s, n):
out = s
while len(out) < n:
out = " " + out
return out
def is_digits(s):
if len(s) == 0:
return False
for c in s:
if c < "0" or c > "9":
return False
return True
def numbers_text(ns):
out = ""
for i in range(0, len(ns)):
if i > 0 and i == len(ns) - 1:
out = out + " and "
elif i > 0:
out = out + ", "
out = out + str(ns[i])
return out
def balances_text(msgs, bal):
out = ""
for a in broker.accounts(msgs):
if out != "":
out = out + ", "
out = out + a + " " + str(bal[a])
return out
def note(msgs, i):
f = broker.first_with_key(msgs, i)
if f == i:
return ""
if broker.same_request(msgs[f], msgs[i]):
return "a retry of message " + str(f + 1)
return "the key of message " + str(f + 1) + ", for another request"
def widths(msgs):
kw = 0
bw = 0
for m in msgs:
if len(m[0]) > kw:
kw = len(m[0])
if len(broker.body_text(m)) > bw:
bw = len(broker.body_text(m))
return [kw, bw]
def right_text(msgs):
want = broker.expected(msgs)
refused = []
for i in range(0, len(msgs)):
if want[1][i] == "refused":
refused = refused + [i + 1]
out = "Right: " + balances_text(msgs, want[0])
if len(refused) == 1:
out = out + "; message " + numbers_text(refused) + " refused"
elif len(refused) > 1:
out = out + "; messages " + numbers_text(refused) + " refused"
return out + "."
def show(msgs, delay, k):
print("Messages (a message whose ack is lost is sent again " + str(delay) + " deliveries later; the window keeps " + str(k) + " keys):")
w = widths(msgs)
for i in range(0, len(msgs)):
m = msgs[i]
line = " " + lpad(str(i + 1), 2) + " " + pad(m[0], w[0]) + " "
n = note(msgs, i)
if n == "":
print(line + broker.body_text(m))
else:
print(line + pad(broker.body_text(m), w[1]) + " " + n)
print(right_text(msgs))
def handler_line(msgs, name, kind, k, seq):
r = broker.handle(kind, k, msgs, seq)
p = broker.problems(msgs, r)
parts = []
if p[0] > 0:
parts = parts + [str(p[0]) + " applied again"]
for x in p[1]:
who = "message " + str(x[0] + 1)
if x[1] == "applied":
parts = parts + [who + " applied under message " + str(x[2] + 1) + "'s key"]
elif x[1] == "told":
parts = parts + [who + " told done with message " + str(x[2] + 1) + "'s answer"]
else:
parts = parts + [who + " wrongly refused"]
if len(parts) == 0 and not broker.balances_right(msgs, r):
parts = parts + ["balances wrong"]
verdict = ""
if len(parts) == 0:
verdict = "right"
if kind == "replay":
verdict = verdict + ", keeping " + str(r[4]) + " keys"
elif kind != "add":
verdict = verdict + ", keeping " + str(r[4]) + " keys with their requests"
else:
for q in parts:
if verdict != "":
verdict = verdict + "; "
verdict = verdict + q
print(" " + pad(name, 10) + balances_text(msgs, r[0]) + ": " + verdict + ".")
def deliver(msgs, delay, k):
print("Acks lost: message numbers from 1 to " + str(len(msgs)) + " such as 2 5 (a number twice loses two), or none.")
t = plain(input("acks lost> "))
if t == "":
print("Cancelled.")
return
lost = []
for m in msgs:
lost = lost + [0]
if t != "none":
w = ""
for c in t + " ":
if c == " ":
if w != "":
if not is_digits(w) or len(w) > 2 or int(w) < 1 or int(w) > len(msgs):
print("Type message numbers from 1 to " + str(len(msgs)) + ", or none.")
return
lost[int(w) - 1] = lost[int(w) - 1] + 1
if lost[int(w) - 1] > 3:
print("At most 3 lost acks for one message.")
return
w = ""
else:
w = w + c
seq = broker.deliveries(msgs, lost, delay)
wd = widths(msgs)
print("Deliveries:")
for d in range(0, len(seq)):
e = seq[d]
m = msgs[e[0]]
tag = ""
if e[2]:
tag = "sent again"
if e[1]:
if tag != "":
tag = tag + ", "
tag = tag + "ack lost"
line = " " + lpad(str(d + 1), 2) + " message " + str(e[0] + 1) + " " + pad(m[0], wd[0]) + " "
if tag == "":
print(line + broker.body_text(m))
else:
print(line + pad(broker.body_text(m), wd[1]) + " " + tag)
print(right_text(msgs))
handler_line(msgs, "add", "add", k, seq)
handler_line(msgs, "replay", "replay", k, seq)
handler_line(msgs, "check", "check", k, seq)
handler_line(msgs, "window " + str(k), "window", k, seq)
def every(msgs, delay):
c = broker.census(msgs, delay)
print("Every choice of which messages lose their first ack: " + str(c[0]) + " runs, each lost one sent again " + str(delay) + " deliveries later, or sooner at the end.")
print(" handler right applied twice reuse mishandled keys kept")
kept = 0
for row in c[1]:
print(" " + pad(row[0], 10) + lpad(str(row[1]), 8) + lpad(str(row[2]), 15) + lpad(str(row[3]), 18) + lpad(str(row[4]), 11))
if row[0] == "check":
kept = row[4]
print(" applied twice: some request was applied more than once.")
print(" reuse mishandled: a key reused for another request was not refused, or a message was wrongly refused.")
last = c[1][len(c[1]) - 1]
print("The first window right on every run keeps " + str(last[4]) + " keys; check keeps " + str(kept) + ". How small a window can be depends on how late a message comes back.")
def add_message(msgs):
if len(msgs) >= 8:
print("8 messages is the most.")
return msgs
key = plain(input("key (up to 8 letters and digits)> "))
if key == "":
print("Cancelled.")
return msgs
if len(key) > 8:
print("Type a key of up to 8 letters and digits.")
return msgs
for c in key:
if not (c >= "a" and c <= "z" or c >= "A" and c <= "Z" or c >= "0" and c <= "9"):
print("Type a key of up to 8 letters and digits.")
return msgs
acct = plain(input("account (up to 10 lowercase letters)> "))
if acct == "":
print("Cancelled.")
return msgs
if len(acct) > 10:
print("Type an account of up to 10 lowercase letters.")
return msgs
for c in acct:
if c < "a" or c > "z":
print("Type an account of up to 10 lowercase letters.")
return msgs
t = plain(input("amount (a whole number from -999 to 999, not 0)> "))
if t == "":
print("Cancelled.")
return msgs
sign = 1
digits = t
if t[0] == "-" or t[0] == "+":
if t[0] == "-":
sign = 0 - 1
digits = t[1:len(t)]
if not is_digits(digits) or len(digits) > 3 or int(digits) == 0:
print("Type a whole number from -999 to 999, not 0.")
return msgs
out = msgs + [[key, acct, sign * int(digits)]]
n = note(out, len(out) - 1)
if n == "":
print("Added message " + str(len(out)) + ": " + key + " " + broker.body_text(out[len(out) - 1]) + ".")
else:
print("Added message " + str(len(out)) + ": " + key + " " + broker.body_text(out[len(out) - 1]) + ", " + n + ".")
return out
def one_to_nine(prompt):
t = plain(input(prompt))
if t == "":
print("Cancelled.")
return 0
if len(t) != 1 or t < "1" or t > "9":
print("Type a number from 1 to 9.")
return 0
return int(t)
msgs = broker.default_messages()
delay = 3
k = 2
print("== Message queue ==")
print("A broker delivers every message at least once: when an ack is lost, it sends the message again. Each message carries an idempotency key.")
running = True
while running:
print("")
print("1) messages 2) deliver 3) every lost ack 4) add a message 5) redelivery 6) window 7) quit")
choice = plain(input("choice> "))
if choice == "1":
show(msgs, delay, k)
elif choice == "2":
deliver(msgs, delay, k)
elif choice == "3":
every(msgs, delay)
elif choice == "4":
msgs = add_message(msgs)
elif choice == "5":
n = one_to_nine("sent again after (1 to 9 deliveries)> ")
if n > 0:
delay = n
print("A message whose ack is lost is now sent again " + str(delay) + " deliveries later.")
elif choice == "6":
n = one_to_nine("window (1 to 9 keys)> ")
if n > 0:
k = n
print("The window now keeps " + str(k) + " keys.")
elif choice == "7":
running = False
else:
print("Pick a number from 1 to 7.")
print("Bye.")
broker.eml
eml# P117 message queue - a broker that delivers every message at least once:
# when a consumer's acknowledgement is lost, the broker sends the message
# again a few deliveries later. As in the corpus case at-least-once-delivery,
# what a duplicate does is up to the handler. Each message carries an
# idempotency key, and as in the corpus case idempotency-key-reuse, a client
# that reuses a key for another request has to be refused, which needs the
# request kept beside the key.
#
# A message is [key, account, amount]; every account starts at 0.
def default_messages():
return [["k1", "alice", 30], ["k2", "bob", 20], ["k3", "alice", 0 - 10], ["k2", "bob", 25], ["k4", "carol", 50], ["k4", "carol", 50], ["k5", "bob", 0 - 5]]
def signed(n):
if n < 0:
return "-" + str(0 - n)
return "+" + str(n)
def body_text(m):
return m[1] + " " + signed(m[2])
def same_request(a, b):
return a[1] == b[1] and a[2] == b[2]
def member(xs, x):
for y in xs:
if y == x:
return True
return False
def accounts(msgs):
[] => out
for m in msgs:
if not member(out, m[1]):
out + [m[1]] => out
return out
def first_with_key(msgs, i):
# the first message carrying message i's key
for j in [0:i]:
if msgs[j][0] == msgs[i][0]:
return j
return i
def expected(msgs):
# each key's first message applied once; a later message with the same
# key and the same request is a retry and is answered done; with another
# request it is refused: [balances, answer per message]
{} => bal
for a in accounts(msgs):
0 => bal[a]
[] => answers
for i in [0:len(msgs) - 1]:
msgs[i] => m
first_with_key(msgs, i) => f
if f == i:
bal[m[1]] + m[2] => bal[m[1]]
answers + ["done"] => answers
elif same_request(msgs[f], m):
answers + ["done"] => answers
else:
answers + ["refused"] => answers
return [bal, answers]
def deliveries(msgs, lost, delay):
# the broker's deliveries, in order, as [message, ack lost, sent before];
# lost[i] is how many of message i's acks are lost. A message whose ack
# is lost is sent again delay deliveries later, or sooner when nothing
# else is waiting.
[] => pending
[] => left
for i in [0:len(msgs) - 1]:
pending + [i] => pending
left + [lost[i]] => left
[] => again
[] => sent
[] => out
0 => d
while len(pending) > 0 or len(again) > 0:
d + 1 => d
0 - 1 => best
for j in [0:len(again) - 1]:
if best == 0 - 1 or again[j][0] < again[best][0]:
j => best
if best != 0 - 1 and (again[best][0] <= d or len(pending) == 0):
again[best][1] => m
[] => rest
for j in [0:len(again) - 1]:
if j != best:
rest + [again[j]] => rest
rest => again
else:
pending[0] => m
pending[1:len(pending)] => pending
left[m] > 0 => lose
if lose:
left[m] - 1 => left[m]
again + [[d + delay, m]] => again
out + [[m, lose, member(sent, m)]] => out
sent + [m] => sent
return out
def handle(kind, k, msgs, seq):
# one handler over the deliveries:
# add applies every delivery;
# replay keeps every key, and answers a key it has seen with the
# first answer, applying nothing;
# check keeps every key with its request: the same request again is
# answered done, another request under it is refused;
# window like check, but keeps only the last k keys it applied.
# [balances, applied again, times applied per message, last answer per
# message, keys kept]
{} => bal
for a in accounts(msgs):
0 => bal[a]
{} => kept
[] => window
[] => applied
0 => again
[] => times
[] => answers
for m in msgs:
times + [0] => times
answers + [""] => answers
for e in seq:
e[0] => i
msgs[i] => m
"apply" => act
if kind == "replay" or kind == "check":
if m[0] in kept:
if same_request(msgs[kept[m[0]]], m):
"repeat" => act
elif kind == "replay":
"told" => act
else:
"refused" => act
elif kind == "window":
for w in window:
if w[0] == m[0]:
if same_request(msgs[w[1]], m):
"repeat" => act
else:
"refused" => act
if act == "apply":
if member(applied, [m[0], m[1], m[2]]):
again + 1 => again
applied + [[m[0], m[1], m[2]]] => applied
bal[m[1]] + m[2] => bal[m[1]]
times[i] + 1 => times[i]
"done" => answers[i]
if kind == "replay" or kind == "check":
i => kept[m[0]]
if kind == "window":
window + [[m[0], i]] => window
if len(window) > k:
window[1:len(window)] => window
elif act == "repeat":
"done" => answers[i]
else:
act => answers[i]
0 => n
if kind == "window":
len(window) => n
else:
len(kept) => n
return [bal, again, times, answers, n]
def problems(msgs, r):
# what went wrong, as [applied again, list of [message, what, first]]
expected(msgs) => want
[] => out
for i in [0:len(msgs) - 1]:
first_with_key(msgs, i) => f
if want[1][i] == "refused":
if r[2][i] > 0:
out + [[i, "applied", f]] => out
elif r[3][i] == "told":
out + [[i, "told", f]] => out
elif r[3][i] == "refused":
out + [[i, "refused", f]] => out
return [r[1], out]
def balances_right(msgs, r):
expected(msgs)[0] => want
for a in accounts(msgs):
if r[0][a] != want[a]:
return False
return True
def right(msgs, r):
problems(msgs, r) => p
return p[0] == 0 and len(p[1]) == 0 and balances_right(msgs, r)
def lost_choices(n):
# every choice of which messages lose their first ack
[[]] => out
for i in [1:n]:
[] => more
for c in out:
more + [c + [0]] => more
more + [c + [1]] => more
more => out
return out
def census(msgs, delay):
# over every choice of lost first acks, for add, replay, check and the
# windows from 1 key up to the first that is right on every run:
# [runs, rows of [name, right, applied twice, reuse mishandled, kept]]
lost_choices(len(msgs)) => choices
[] => runs
for c in choices:
runs + [deliveries(msgs, c, delay)] => runs
[] => rows
["add", "replay", "check"] => kinds
0 => k
True => going
while going:
if len(kinds) > 0:
kinds[0] => kind
kinds[1:len(kinds)] => kinds
kind => name
else:
"window" => kind
k + 1 => k
"window " + str(k) => name
0 => good
0 => twice
0 => reuse
0 => most
for seq in runs:
handle(kind, k, msgs, seq) => r
problems(msgs, r) => p
if p[0] == 0 and len(p[1]) == 0 and balances_right(msgs, r):
good + 1 => good
if p[0] > 0:
twice + 1 => twice
if len(p[1]) > 0:
reuse + 1 => reuse
if r[4] > most:
r[4] => most
rows + [[name, good, twice, reuse, most]] => rows
if kind == "window" and good == len(runs):
False => going
return [len(runs), rows]
Python projection (broker.py)
def default_messages():
return [["k1", "alice", 30], ["k2", "bob", 20], ["k3", "alice", 0 - 10], ["k2", "bob", 25], ["k4", "carol", 50], ["k4", "carol", 50], ["k5", "bob", 0 - 5]]
def signed(n):
if n < 0:
return "-" + str(0 - n)
return "+" + str(n)
def body_text(m):
return m[1] + " " + signed(m[2])
def same_request(a, b):
return a[1] == b[1] and a[2] == b[2]
def member(xs, x):
for y in xs:
if y == x:
return True
return False
def accounts(msgs):
out = []
for m in msgs:
if not member(out, m[1]):
out = out + [m[1]]
return out
def first_with_key(msgs, i):
for j in range(0, i+1):
if msgs[j][0] == msgs[i][0]:
return j
return i
def expected(msgs):
bal = {}
for a in accounts(msgs):
bal[a] = 0
answers = []
for i in range(0, len(msgs)):
m = msgs[i]
f = first_with_key(msgs, i)
if f == i:
bal[m[1]] = bal[m[1]] + m[2]
answers = answers + ["done"]
elif same_request(msgs[f], m):
answers = answers + ["done"]
else:
answers = answers + ["refused"]
return [bal, answers]
def deliveries(msgs, lost, delay):
pending = []
left = []
for i in range(0, len(msgs)):
pending = pending + [i]
left = left + [lost[i]]
again = []
sent = []
out = []
d = 0
while len(pending) > 0 or len(again) > 0:
d = d + 1
best = 0 - 1
for j in range(0, len(again)):
if best == 0 - 1 or again[j][0] < again[best][0]:
best = j
if best != 0 - 1 and (again[best][0] <= d or len(pending) == 0):
m = again[best][1]
rest = []
for j in range(0, len(again)):
if j != best:
rest = rest + [again[j]]
again = rest
else:
m = pending[0]
pending = pending[1:len(pending)]
lose = left[m] > 0
if lose:
left[m] = left[m] - 1
again = again + [[d + delay, m]]
out = out + [[m, lose, member(sent, m)]]
sent = sent + [m]
return out
def handle(kind, k, msgs, seq):
bal = {}
for a in accounts(msgs):
bal[a] = 0
kept = {}
window = []
applied = []
again = 0
times = []
answers = []
for m in msgs:
times = times + [0]
answers = answers + [""]
for e in seq:
i = e[0]
m = msgs[i]
act = "apply"
if kind == "replay" or kind == "check":
if m[0] in kept:
if same_request(msgs[kept[m[0]]], m):
act = "repeat"
elif kind == "replay":
act = "told"
else:
act = "refused"
elif kind == "window":
for w in window:
if w[0] == m[0]:
if same_request(msgs[w[1]], m):
act = "repeat"
else:
act = "refused"
if act == "apply":
if member(applied, [m[0], m[1], m[2]]):
again = again + 1
applied = applied + [[m[0], m[1], m[2]]]
bal[m[1]] = bal[m[1]] + m[2]
times[i] = times[i] + 1
answers[i] = "done"
if kind == "replay" or kind == "check":
kept[m[0]] = i
if kind == "window":
window = window + [[m[0], i]]
if len(window) > k:
window = window[1:len(window)]
elif act == "repeat":
answers[i] = "done"
else:
answers[i] = act
n = 0
if kind == "window":
n = len(window)
else:
n = len(kept)
return [bal, again, times, answers, n]
def problems(msgs, r):
want = expected(msgs)
out = []
for i in range(0, len(msgs)):
f = first_with_key(msgs, i)
if want[1][i] == "refused":
if r[2][i] > 0:
out = out + [[i, "applied", f]]
elif r[3][i] == "told":
out = out + [[i, "told", f]]
elif r[3][i] == "refused":
out = out + [[i, "refused", f]]
return [r[1], out]
def balances_right(msgs, r):
want = expected(msgs)[0]
for a in accounts(msgs):
if r[0][a] != want[a]:
return False
return True
def right(msgs, r):
p = problems(msgs, r)
return p[0] == 0 and len(p[1]) == 0 and balances_right(msgs, r)
def lost_choices(n):
out = [[]]
for i in range(1, n+1):
more = []
for c in out:
more = more + [c + [0]]
more = more + [c + [1]]
out = more
return out
def census(msgs, delay):
choices = lost_choices(len(msgs))
runs = []
for c in choices:
runs = runs + [deliveries(msgs, c, delay)]
rows = []
kinds = ["add", "replay", "check"]
k = 0
going = True
while going:
if len(kinds) > 0:
kind = kinds[0]
kinds = kinds[1:len(kinds)]
name = kind
else:
kind = "window"
k = k + 1
name = "window " + str(k)
good = 0
twice = 0
reuse = 0
most = 0
for seq in runs:
r = handle(kind, k, msgs, seq)
p = problems(msgs, r)
if p[0] == 0 and len(p[1]) == 0 and balances_right(msgs, r):
good = good + 1
if p[0] > 0:
twice = twice + 1
if len(p[1]) > 0:
reuse = reuse + 1
if r[4] > most:
most = r[4]
rows = rows + [[name, good, twice, reuse, most]]
if kind == "window" and good == len(runs):
going = False
return [len(runs), rows]