Message Log Reader

Track: Protocols and sockets. 6 stages. Languages: Go.

Some systems, such as Kafka, keep messages in a log: a long list where each new message goes at the end and nothing is ever changed. Each message has a position: the first one is 0, the next is 1, and so on. A program that reads the log works through it one message at a time. It also tells the server how far it got, so that if it stops or crashes, it can carry on from that place later. Build that reader.

The log server is given. It speaks a tiny text protocol over a network connection. You send one line, and it answers with one line. Every line ends with \n. The first number in each line says which log you mean. In every stage there is only one log, number 0.

you send          server answers                    meaning
FETCH 0 5         MSG 5 hello  or  EMPTY  or  REVOKED   give me the message at position 5
COMMIT 0 6        OK                                save my place: next time, start at 6
COMMITTED 0       6                                 where is my saved place? (0 if none yet)

EMPTY means there is no message at that position: you have read everything. REVOKED means the server is taking this log away from you, so another reader can take over.

One rule matters more than any other: save your place only after the work on a message is finished. If you save position 6 and then crash while working on message 5, nobody will ever do message 5.

Implement these, using your language's own names for them:

New(dial, partitions) creates a reader. dial is a function we give you. Each call opens one new connection to the server, or fails. Always use it, never connect to the server yourself. partitions is how many logs to read. It is always 1 here, and that log is number 0.
Run(process) reads the log. Ask COMMITTED 0 for your saved place. Then loop: FETCH the message at that position, call process with it, then send COMMIT with the next position. Return when you get EMPTY, or after a REVOKED, or when the tester tells you to stop. If the connection breaks, open a new one with dial and carry on from your saved place.
Metrics() reports two numbers you count yourself: how many MSG answers you got, and how many COMMIT lines the server answered with OK.

Standard library only.

The tester puts a small relay between your reader and the server, so it can break a connection on purpose. It also reads the server's own record of every message it sent and every place you saved, so it knows exactly what happened, whatever your code reports. For help with the ideas, read the notes.

Where it is used

Apache Kafka readers save their position in the log on the server, so a reader that restarts, or a new reader that takes over, carries on where the last one stopped.
Amazon Kinesis apps built with the Kinesis Client Library save a checkpoint for each part of the stream in a database table, for the same reason.

The order is the whole point. Save your place before the work, and a crash loses messages. Save it after, and a crash only repeats some. This challenge asks for the second one.

Stages

ko1: read every message in order and save your place (easy)

The log has 8 messages, at positions 0 to 7. Read them all, in order, and save your place after each one. You have up to 1.5 seconds.

COMMITTED 0   -> 0              (no saved place yet, start at 0)
FETCH 0 0     -> MSG 0 p0-m0    process it, then
COMMIT 0 1    -> OK             (next time, start at 1)
FETCH 0 1     -> MSG 1 p0-m1    ... and so on up to position 7
COMMIT 0 8    -> OK
FETCH 0 8     -> EMPTY          (all read, return)

The tester will verify: every message from 0 to 7 was read with none skipped, your saved place never moved back, and it ended at 8.

ko2: save your place only after the work is done (medium)

The same log of 8 messages. This time the tester also checks the order of your steps: process must finish a message before you save a place past it. To see the order, the tester makes process slow on one message: it takes 200 ms on message 3. Your COMMIT 0 4 must come after that.

FETCH 0 3     -> MSG 3 p0-m3
process(3), then COMMIT 0 4     (correct)
COMMIT 0 4, then process(3)     (wrong: if you crash in between, message 3 is lost)

The tester will verify: everything from ko1, your saved place was never ahead of the messages process had finished, and COMMIT 0 4 came only after process finished message 3.

ko3: read messages quickly (L2, hard)

Keep Run fast. The tester gives it a log of 20,000 messages with no real network: dial returns a fake connection that answers at once. Every message must still be fetched, handed to process, and saved with a real COMMIT line. Run must return when it gets EMPTY. Example: if our solution takes 100 ms, the limit is 150 ms. Your solution takes 140 ms, so it passes. In Go, your code may also make at most twice as many memory allocations as ours.

The tester will verify: your time is at most 1.5 times our solution's time, and every message really reached process.

ko4: carry on from your saved place when the connection breaks (L3, hard)

The log has 12 messages, and you have up to 2.5 seconds. When you save place 4, the server answers OK and then closes connection #1. The tester then breaks connection #2 as soon as it opens.

connection #1:  messages 0 to 3, COMMIT 0 4 -> OK, then closed
dial         -> connection #2, broken at once
dial         -> connection #3              (open a new one, don't give up)
COMMITTED 0  -> 4                          (carry on from 4, not from 0)
FETCH 0 4    -> MSG 4 p0-m4 ... and so on up to position 11

The tester will verify: all 12 messages reached process exactly once each, with none skipped and none repeated, your saved place never moved back, and the tester broke exactly 1 connection.

ko5: save your place and stop when the log is taken away (L3, hard)

The log has 8 messages, and you have up to 2.5 seconds. When you ask for position 4, the server answers REVOKED. Save your place, then stop reading. The log now belongs to another reader.

FETCH 0 3     -> MSG 3 p0-m3, process it, COMMIT 0 4 -> OK
FETCH 0 4     -> REVOKED
COMMIT 0 4    -> OK             (save your place, then return)

The tester will verify: the server sent REVOKED once, and no message at position 4 or later reached process.

ko6: report your own numbers correctly (L4, medium)

The same log as ko1: 8 messages, no broken connections. At the end, the tester compares your Metrics() with the server's own record. Count a message only when the server answered MSG. Count a save only when the server answered OK to your COMMIT. With one save after each message, both numbers are 8 here.

server saw: 8 messages sent, saved place 8
Metrics() -> Fetched 8, Committed 8     (correct)
Metrics() -> Fetched 8, Committed 0     (wrong: never counted the saves)

The tester will verify: your message count and your save count match the server's record exactly.

Rules

Standard library only.
Use the dial function you are given. Never connect to the server address yourself.
Call process first, then save your place past that message, never the other way round.
A broken connection is a reason to open a new one and carry on from your saved place. It is not a reason to stop, and not a reason to start again from 0.
After a REVOKED, save your place and stop reading.

Read the Message Log Reader notes in the wiki