Q5 - The Fleet Message Log

Scenario: Google runs a monitoring service for a big fleet of machines. Each machine has a stable, human-readable name (VM A, VM B, and so on) and streams short status messages to the service. Every message carries three things: the machine's name, some text, and a timestamp.

They show up looking like this:

VM A, online, time:5
VM B, ready,  time:2
VM C, up,     time:1

Part 1: Messages stream in one at a time. Build something that takes them in and can hand back, for any machine, that machine's messages in timestamp order.

For Part 1 we will make one reasonable assumption (and this is exactly the kind of thing you would confirm with your interviewer): each machine's messages arrive in increasing timestamp order. So VM A at time 5 arrives before VM A at time 8. Different machines can still interleave any which way.

MESSAGES (in arrival order):
   VM A, online,  time:5
   VM B, ready,   time:2
   VM C, up,      time:1
   VM A, healthy, time:8
   VM B, warn,    time:4

HISTORY per machine (timestamp order):
   VM A  ->  [ (5, online), (8, healthy) ]
   VM B  ->  [ (2, ready), (4, warn) ]
   VM C  ->  [ (1, up) ]

Part 1: Model It First

Strip away the story and the shape falls out. You have messages keyed by machine name, and for each machine you want them in time order. That is a map from name to an ordered list of messages.

• The key is the machine name.
• The value is a growable list of (timestamp, text) pairs, kept in timestamp order.

Because we assumed each machine's messages already arrive in increasing time order, ingesting one message is just: look up the machine's list (create it if this is the first time we have heard from it) and append. Handing back a machine's history is just returning its list.

A natural second query to offer, and one worth mentioning out loud, is "what is the latest message from machine X?" With the list kept in time order that is simply the last element. Keep that in mind, because it is the hinge the second part turns on.

# record: O(1) append   |   history: O(k) for k messages
class FleetLog:
    def __init__(self):
        self.by_machine = {}                 # name -> list of (timestamp, text), in time order

    def record(self, name, text, ts):
        self.by_machine.setdefault(name, []).append((ts, text))

    def history(self, name):
        return list(self.by_machine.get(name, []))

That is the warm-up. If it feels too easy, that is because the interviewer has not dropped the other shoe yet.

Part 2: When the Clocks Lie

Here is the wrinkle. The fleet has a problem: a machine's messages do not always arrive in timestamp order. Clocks drift, packets get retried, and every so often you record a message from VM A at time 5, and the very next thing VM A sends you is stamped time 3, earlier than one you already stored.

When that happens, the rule is: trust the newer arrival, and stop trusting everything that came after it. Concretely, when a message from a machine arrives with timestamp t:

Drop every message you are currently holding for that machine whose timestamp is greater than t, then store the new message at t.

So a message that steps backward in time is treated as a correction. It rewinds that machine's history to time t and picks up from there. Other machines are untouched. Walk through two cases before looking at any code.

MESSAGES from VM A (arrival order):
   boot       time:1
   sync       time:4
   heartbeat  time:7
   resync     time:3     <-- steps BACK (3 < 7)

STORED HISTORY for VM A, step by step:
   after boot@1        [ (1,boot) ]
   after sync@4        [ (1,boot), (4,sync) ]
   after heartbeat@7   [ (1,boot), (4,sync), (7,heartbeat) ]
   after resync@3      drop everything with time > 3  (that is 4 and 7),
                       then add 3
                       [ (1,boot), (3,resync) ]

RESULT:  VM A  ->  [ (1, boot), (3, resync) ]

Notice that one backward message removed two stored messages, not one. Both sync@4 and heartbeat@7 are later than 3, so both are now suspect and both go.

MESSAGES (arrival order):
   VM A, wake,    time:2
   VM B, wake,    time:2
   VM A, scan,    time:5
   VM A, rescan,  time:3     <-- VM A steps back (3 < 5)
   VM B, idle,    time:6
   VM A, idle,    time:8

STORED HISTORY, step by step:
   A wake@2      A: [ (2,wake) ]
   B wake@2      B: [ (2,wake) ]
   A scan@5      A: [ (2,wake), (5,scan) ]
   A rescan@3    A: drop time > 3 (removes 5,scan), add 3   -> [ (2,wake), (3,rescan) ]
   B idle@6      B: [ (2,wake), (6,idle) ]        (VM B is completely unaffected)
   A idle@8      A: [ (2,wake), (3,rescan), (8,idle) ]

RESULT:
   VM A  ->  [ (2, wake), (3, rescan), (8, idle) ]
   VM B  ->  [ (2, wake), (6, idle) ]

Two things to take from Example 3. First, VM A stepping back never touches VM B; each machine's timeline is its own. Second, after the rewind to time 3, VM A keeps going and idle@8 appends normally on top. A correction is not the end of that machine's history, just a rewind.

Step 1: The list stays sorted, so it is a stack

For a single machine, the list you are holding is always sorted by timestamp, smallest at the bottom, largest on top. It starts that way and stays that way, so the newest message is always the one sitting on top.

VM A after boot@1, sync@4, heartbeat@7  (sorted, newest on top):

   top ->  (7, heartbeat)
           (4, sync)
           (1, boot)

Step 2: A backward message pops the too-new tops, then pushes

Because the list is sorted, "drop every message with a timestamp greater than t" is not a scan of the whole list. The messages greater than t are exactly the ones on top, so you pop while the top's timestamp is greater than t, then push the new message. That is a monotonic stack, one per machine, and every message is pushed once and popped at most once.

resync@3 arrives  (3 < top 7):
   pop 7   (7 > 3)      ->  [ 1, 4 ]
   pop 4   (4 > 3)      ->  [ 1 ]
   top 1 <= 3, so stop  ->  push 3  ->  [ 1, 3 ]

That pop-then-push is the whole record method: one while that clears the too-new tops, then a single append.

stack = by_machine[name]                 # this machine's messages, timestamps increasing
while stack and stack[-1][0] > ts:       # a backward step: pop everything newer than ts
    stack.pop()
stack.append((ts, text))                 # then push the new message on top

Watch VM A's stack handle Example 2 frame by frame: the normal messages push on top, then resync@3 pops the too-new ones before it lands:

1 / 7

# record: amortized O(1)   |   history: O(k)
class FleetLog:
    def __init__(self):
        self.by_machine = {}                  # name -> stack of (timestamp, text), increasing

    def record(self, name, text, ts):
        stack = self.by_machine.setdefault(name, [])
        while stack and stack[-1][0] > ts:    # backward step: drop everything after ts
            stack.pop()
        stack.append((ts, text))

    def history(self, name):
        return list(self.by_machine.get(name, []))

Part 3: The Live Dashboard

The last ask is the screen the on-call engineer actually watches. The service powers a dashboard that shows the k most recent messages across the whole fleet, newest first, and it refreshes as the stream keeps flowing. Add a recentFeed(k) that returns them.

One thing to pin down out loud before you build it: if two machines have a message at the very same timestamp, which one counts as "more recent"? Any consistent rule works. We will say the machine whose name comes first alphabetically wins, so the answer is deterministic.

FLEET STATE (from Example 3, after the corrections have settled):
   VM A  ->  [ (2, wake), (3, rescan), (8, idle) ]
   VM B  ->  [ (2, wake), (6, idle) ]

recentFeed(3)  ->  [ (8, VM A, idle),
                     (6, VM B, idle),
                     (3, VM A, rescan) ]

Two things worth seeing here:
  - VM A's scan@5 is nowhere in the feed. Part 2 already threw it away when
    rescan@3 arrived, so a corrected-away message can never leak onto the
    dashboard. You get that for free.
  - ask for recentFeed(5) and the tail is wake@2 from BOTH machines, and the
    tie at time 2 breaks by name, so VM A lands just ahead of VM B.

Step 1: The naive feed, and why it is wasteful

The obvious version: dump every trusted message from every machine into one list, sort by timestamp, and take the first k. It works, but it is O(n log n) in the total number of stored messages, and you throw away almost all of that sorting just to show k rows.

pool everything, sort, take k = 3:

   [ 2:A wake, 3:A rescan, 8:A idle, 2:B wake, 6:B idle ]
   sort by time, newest first -> [ 8, 6, 3, 2, 2 ] -> take 3 -> [ 8, 6, 3 ]

Step 2: The freshest message is always a stack top

Now use what Part 2 handed you. Each machine's stack is already sorted, with its newest message right on top. So the freshest message in the entire fleet has to be the top of one of those stacks, nothing buried in the middle can beat it. Take it, and the next-freshest is again a top: either another machine's top, or the message directly beneath the one you just took.

each machine's stack, newest on top:

   VM A top = (8, idle)      VM B top = (6, idle)

the newest message anywhere has to be one of these tops.

Step 3: A k-way merge with a max-heap

Keep a max-heap holding one "cursor" per machine, each starting at that machine's top. Pop the newest, add it to the feed, then push that machine's next-older message back in, and repeat k times. Only the k messages you actually show are ever touched.

seed the heap with each machine's top, then pop k times:

   heap { 8:A, 6:B }   pop 8:A  ->  feed [ 8:A ]       push A's next, 3
   heap { 6:B, 3:A }   pop 6:B  ->  feed [ 8, 6 ]      push B's next, 2
   heap { 3:A, 2:B }   pop 3:A  ->  feed [ 8, 6, 3 ]   (k = 3, stop)

In code it is a small loop: pop the newest cursor, record it, then push that machine's next-older message, k times.

# heap holds one cursor per machine (negated ts, so the newest pops first)
while heap and len(feed) < k:
    neg_ts, name, idx = heapq.heappop(heap)      # newest message across the fleet
    feed.append(by_machine[name][idx])
    if idx > 0:                                  # walk one message older on this machine
        heapq.heappush(heap, (-by_machine[name][idx - 1][0], name, idx - 1))

Step through the merge. Each machine keeps a cursor into its stack; the heap always holds the current tops, and popping the max builds the feed newest-first:

1 / 5

# import heapq
# add to FleetLog:
def recent_feed(self, k):
    # one cursor per machine at its top; pop newest (max ts, then name ascending)
    heap = [(-stack[-1][0], name, len(stack) - 1)
            for name, stack in self.by_machine.items() if stack]
    heapq.heapify(heap)
    out = []
    while heap and len(out) < k:
        neg_ts, name, idx = heapq.heappop(heap)
        ts, text = self.by_machine[name][idx]
        out.append((ts, name, text))
        if idx > 0:                                  # walk one message older on this machine
            heapq.heappush(heap, (-self.by_machine[name][idx - 1][0], name, idx - 1))
    return out

The Full Solution

Here is the whole FleetLog in one piece, all three operations together, with the time and space cost noted right above each method.

import heapq

class FleetLog:
    def __init__(self):
        self.by_machine = {}                          # name -> stack of (timestamp, text), increasing

    # amortized O(1) time | O(1) extra space
    def record(self, name, text, ts):
        stack = self.by_machine.setdefault(name, [])
        while stack and stack[-1][0] > ts:            # backward step: drop everything after ts
            stack.pop()
        stack.append((ts, text))

    # O(k) time | O(k) space   (k = messages this machine holds)
    def history(self, name):
        return list(self.by_machine.get(name, []))

    # O(M + k log M) time | O(M) space   (M = machines that hold messages)
    def recent_feed(self, k):
        heap = [(-stack[-1][0], name, len(stack) - 1)
                for name, stack in self.by_machine.items() if stack]
        heapq.heapify(heap)
        out = []
        while heap and len(out) < k:
            neg_ts, name, idx = heapq.heappop(heap)
            ts, text = self.by_machine[name][idx]
            out.append((ts, name, text))
            if idx > 0:                               # walk one message older on this machine
                heapq.heappush(heap, (-self.by_machine[name][idx - 1][0], name, idx - 1))
        return out

Complexity Analysis

• record: amortized O(1). A single backward message can pop many entries, so one call is O(number popped) in the worst case. But every message is pushed exactly once and popped at most once over the whole run, so ingesting a stream of n messages is O(n) total. Averaged out, each call is constant.
• history(name): O(k) for a machine holding k messages, since you copy them out (or O(1) if you hand back a reference and trust the caller).
• Space: O(m) for m messages currently stored across the fleet. Corrections only ever shrink what you keep.
• recentFeed(k): O(machines + k log machines). You seed the heap with each machine's newest message, then pop k times, each pop costing O(log machines). That beats gathering every stored message into one list and sorting it, which is O(n log n).

The naive alternative is worth naming out loud so the interviewer knows you see it: keep each machine's messages in a plain list and, on a backward message, walk the list and delete everything with a larger timestamp. That is O(k) per correction and can degrade to O(n^2) over a stream. The stack gets you the same result in amortized O(1) because the entries you need to delete are always the ones on top.

Final Notes

1. The model is most of the answer. A map from machine name to an ordered list of messages is the whole of Part 1. Getting to "group by key, keep each group in time order" quickly, and saying why, is what the first half is really checking.

2. A backward timestamp is a rewind, not a delete-one. The rule removes every message after time t, which can be several at once (Example 2 drops two). Read that carefully in the room; it is the detail people skim past and get wrong.

3. "Delete everything after t" on a time-ordered list is a stack pop. Because each machine's list stays sorted, the messages to remove are exactly the ones on top, so you pop while the top is too large, then push. That is a per-machine monotonic stack, and spotting it is the leap that makes Part 2 clean and gives you amortized O(1).

4. Ask first, code second. The prompt hands you a few example lines and nothing else on purpose. The candidates who do well settle the stream-vs-batch, duplicate-name, and output-shape questions up front, and flag the one genuinely ambiguous case (what if a new message ties the timestamp of one already stored?) instead of silently guessing.

5. A sorted-per-machine structure turns the fleet feed into a k-way merge. Every machine's newest message already sits on top of its stack, so the most recent message anywhere is one of those tops. A max-heap with one cursor per machine gives you the k most recent in O(machines + k log machines), with no full sort, and because Part 2 already dropped the corrected-away messages, the feed cannot show a ghost.

The reason a question this innocent shows up at Google is that it separates people who can only run a known algorithm from people who can build a small system from a fuzzy description and keep reshaping it as each new part lands. Get comfortable talking through the model out loud, and the stack and the heap fall out of it on their own. 🚀