Riemann’s consistency trade-offs

Riemann is built for high-throughput, partial-harvest event processing: millions of events per second per node, where dropping some data beats waiting on coordination. A monitoring system can tolerate losing half of its request-latency samples—that’s just a lossy histogram. Timely, approximate processing beats exact, delayed results. If one node’s clock is skewed, it’s better to process an event “soonish” than to block indefinitely waiting for it to catch up.

Events themselves carry no synchronization or cross-event relationship. They are immutable values with a total order, even if any single server sees only a slice of the whole stream. In a sense, events form a transaction log—except the semantics of those transactions are defined by the stream configuration, not by the events themselves.

The current design is only trivially distributed. Clients send to servers; servers can act as clients. The protocol acknowledges each received event synchronously, but what that acknowledgement means is deliberately loose: it might confirm a durable disk write, a drop into an in-memory queue, or a queued write that also triggers an email to your pager. Those guarantees exist per server, not across a cluster. Eventually, Riemann needs to keep working when the network splits.

Fuck it, no coordination

The “no coordination” model accepts degraded harvest and low latency for data that you can afford to lose. A pragmatic approach is to carpet-bomb every Riemann server with each event, setting a write-replica threshold as a tunable. Servers will diverge in their views of the world depending on partition topology and duration—but for monitoring data, that can be fine.

Some events, however, demand stronger coordination. Suppose three Riemann servers guard a datacenter. Exactly one must send the “web server is down” email. That requires both liveness—someone acts within five seconds—and safety—only one actor fires. In the face of arbitrary partitions, these constraints essentially violate CAP. Consider a node that sends the email, then dies before informing its peers. The survivors have no option but to resend, duplicating the alert.

For most failure modes, duplicates are the right trade-off. Paging twice is far better than not paging at all. And for bounded failures—up to floor(n/2) node losses, or partitions that preserve a fully connected quorum—protocols like 2PC or Paxos can still deliver once. You can detect the failure modes that would force duplicate side effects and warn the recipient: “I’ve got split brain, I’ll call twice maybe.”

(streams
  (where (state "critical")
    ; This is unsynchronized and proceeds on all nodes concurrently
    #(prn "Uh oh, this thing's broken!" %)

    (master
      ; Any events inside master are executed on exactly one node if 
      ; quorum is preserved, or maybe multiple hosts if a node fails before
      ; acking.
      (email "[email protected]"))))

This is most valuable when clients can reach a majority of servers and want a guarantee that their event was accepted. A weaker variant—“try to prevent all connected peers from sending this event within a time window”—fits scenarios where clients stay partitioned with their local server, like one Riemann per aggregation switch or datacenter.

None of this ensures every node shares the same picture of the events leading up to a failure. That would require full coordination over the stream’s ordering, imposing nontrivial synchronization costs. Explicit causal consistency could help, but you’d need a way to express causal relationships between arbitrary stream functions. In practice, this may not matter: on quorum loss Riemann can wake someone up, and once the partition heals, nodes quickly converge on “that service still isn’t checking in.”

Consistency for historical stores

The open question is whether there’s a middle path—events that don’t need the overhead of 2PC or Paxos per event, but still need some distributed consistency. HAT offers reasonably strong consistency for an AP system, but sacrifices liveness. That’s probably unsuitable for Riemann’s real-time alerting. However, HAT-style semantics might fit historical stores or distributed multi-event transactions—neither of which exist in Riemann today.

Riemann’s event model may make reconciliation simple. Events are values, and well-behaved clients impose a total order by host, service, and timestamp. Reconstructing any linear slice of the stream can be done eventually consistently; merging divergent histories is just the set union of received events.

Derived events complicate that. If a partition splits two servers measuring request throughput, each sees half the events and emits rate streams with half the original metric. Writing those derived events to a historical store shows only half the real throughput. One option is to log only raw events and reconstruct derived ones by replaying the merged log—compute the rate at noon by applying events from 11:55 to 12:00 to the rate stream. Another would make rate streams transactional, but that risks breaking the liveness guarantees that make Riemann useful in the first place.