Delivering billions of messages strictly once

The only requirement for all data transmission systems is that data cannot be lost . Data can usually arrive late or can be requested again, but it should never be lost.

To meet this requirement, most distributed systems guarantee at least one-time delivery . Techniques for providing “at least one-time delivery” usually come down to “retries, retries, and retries . ” You never consider a message delivered until you receive a clear confirmation from the client.

But as a user , at least a one-time delivery is not quite what I want. I want messages to be deliveredonce . And only once.

Unfortunately, to achieve something close to exactly one-time delivery , an impenetrable design is needed . The architecture provides that each case of failure must be carefully considered - it will not be possible to simply register it as part of the existing implementation after the fact of the failure. And even then, it is almost impossible to implement a system in which messages are delivered only once.

Over the past three months, we have developed a completely new deduplication system and as close as possible to exactly one-time delivery, faced with a large number of various failures.

The new system is able to track 100 times more messages than the old one, it differs from it by increased reliability and lower cost. Here's how we did it.

Problem

Most Segment internal systems gracefully handle failures with retries, posting messages, locking, and two-stage commits . But there is one notable exception: clients who send data directly to our public API .

Clients (especially mobile ones) often experience communication interruptions when they can send data, but skip the response from our API.

Imagine that you are taking a bus and have booked a room from the HotelTonight app on your iPhone. The application starts uploading data to Segment servers, but the bus suddenly enters the tunnel and you lose connection. Some of the events you sent are already processed, but the client will never receive a response from the server.

In such cases, clients repeat sending the same events to the Segment API despite the fact that the server has technically already received exactly the same messages earlier.

Judging by the statistics of our server, approximately 0.6% of the events received over the past four weeks are repeated messages that we have already received.

The level of errors may seem insignificant. But for an e-commerce application that generates billions of dollars in revenue, a 0.6% difference could mean a difference between profit and loss of millions of dollars.

Message deduplication

So, we understand the essence of the problem - we need to remove duplicate messages that are sent to the API. But how to do that?

At the theoretical level, the high-level API of any deduplication system seems simple. In Python ( aka pseudo-pseudo-code ), we can represent it as follows:

def dedupe(stream):
  for message in stream:
    if has_seen(message.id): 
      discard(message)
    else:
      publish_and_commit(message)

For each message in the stream, it is first checked whether such a message has been encountered before (by its unique identifier). If met, then discard it. If you have not met, then re-release the message and atomically carry out the transfer.

In order not to store all messages permanently, a “deduplication window” operates, which is defined as the storage time of our keys until their expiration date. If messages do not fit the window, they are considered obsolete. We want to ensure that only one message with this ID is sent in the window.

This behavior is easy to describe, but there are two details that require special attention: read / write performance and accuracy .

We want the system to deduplicate billions of events in our data stream — and do it simultaneously with low latency and cost-effective.

Moreover, we want to make sure that information about registered events is reliably stored, so that we can restore it in the event of a failure, and that there will never be repeated messages in the display.

Architecture

To achieve this, we have created a “two-stage” architecture that reads data from Kafka and removes duplicate events already recorded in the four-week window.


High-Level Deduplication Architecture

Kafka Topology

To understand how this architecture works, first take a look at the Kafka flow topology . All incoming API calls are split into separate messages and clearly express the Kafka input section.

First, each incoming message is marked unique messageId, which is generated on the client side. This is usually UUIDv4 (although we are considering switching to ksuid ). If the client does not report messageId, then we automatically assign it at the API level.

We do not use vector clocks or serial numbers, because we do not want to complicate the client side. Using UUIDs makes it easy for anyone to send data to our API, because almost every major programming language supports it.

{
  "messageId": "ajs-65707fcf61352427e8f1666f0e7f6090",
  "anonymousId": "e7bd0e18-57e9-4ef4-928a-4ccc0b189d18",
  "timestamp": "2017-06-26T14:38:23.264Z",
  "type": "page"
}

Individual messages are logged in Kafka for durability and repeatability. They are distributed by messageId, so that we can be sure that the same messageIdone will always go to the same handler.

This is an important detail when it comes to data processing. Instead of searching in the central database of the key among hundreds of billions of messages, we were able to narrow the search space by orders of magnitude simply by redirecting the search query to a specific section.

Deduplication Worker is a Go program that reads Kafka input sections. She is responsible for reading messages, checking for duplicates, and if the messages are new, for sending them to the Kafka output topic.

In our experience, Kafka Workers and Topologies are extremely easy to manage. We no longer have many large Memcached instances that require failover replicas. Instead, we used the built-in RocksDB databases , which do not need coordination at all and give us persistent storage at an exceptionally low price. Now more about this.

Worker RocksDB

Each worker stores the local RocksDB database on his local EBS hard drive. RocksDB is an integrated key-value repository developed by Facebook and is optimized for extremely high performance.

Whenever an event is retrieved from the input partitions, the consumer requests RocksDB to see if such an event has occurred before messageId.

If the message is not in RocksDB, we add the key to the database and then publish the message in the output sections of Kafka.

If the message is already in RocksDB, the worker simply does not publish it in the output sections and updates the input section with a notification that he processed the message.

Performance

In order to achieve high performance from our database, we must correspond to three types of queries for each event processed:

1. Detecting the existence of random keys that arrive at the input, but are unlikely to be stored in our database. They can be located anywhere in the key space.
2. Record new keys with high performance.
3. Declaring old keys that did not fall into our “deduplication window”.

As a result, we must continuously scan the entire database, add new keys, and obsolete old keys. And ideally, this should happen within the framework of the previous data model.


Our database must satisfy three very different types of queries.

Generally speaking, the bulk of the performance gain comes from the database performance - so it makes sense to understand the RocksDB device, which is why it does the job so well.

RocksDB is a log-structured tree (LSM tree) , that is, it continuously adds new keys to the write-ahead log on disk, and also stores sorted keys in memory as part of memtable .


Keys are sorted in memory as part of memtable.

Writing keys is an extremely fast process. New items are written directly to disk by adding to the log (for direct saving and recovery in the event of a failure), and data records are sorted in memory to provide quick search and batch recording.

Whenever a sufficient number of entries arrive in a memtable , they are saved to disk as SSTable (sorted string table). Since the rows have already been sorted in memory, they can be directly flushed to disk.


The current state of memtable is flushed to disk as SSTable at level zero (Level 0)

Here is an example of such a reset from our working logs:

[JOB 40] Syncing log #655020
[default] [JOB 40] Flushing memtable with next log file: 655022
[default] [JOB 40] Level-0 flush table #655023: started
[default] [JOB 40] Level-0 flush table #655023: 15153564 bytes OK
[JOB 40] Try to delete WAL files size 12238598, prev total WAL file size 24346413, number of live WAL files 3.

Each SSTable table remains unchanged - after it is created, it never changes - thanks to this, the writing of new keys occurs so quickly. No need to update files, and the record does not spawn new records. Instead, multiple SSTable tables at the same “level” merge into a single file during the out-of-band compression phase.



When individual SSTable tables are compacted from one level, their keys merge together, and then the new file is transferred to a higher level. You can find examples of such seals in our working logs. In this case, the process 41 compresses the four zero-level files and combines them into a larger first-level file. When summarization is complete, the combined SSTable tables become the final set of database records, and the old SSTable tables are unchained.

/data/dedupe.db$ head -1000 LOG | grep "JOB 41"
[JOB 41] Compacting 4@0 + 4@1 files to L1, score 1.00
[default] [JOB 41] Generated table #655024: 1550991 keys, 69310820 bytes
[default] [JOB 41] Generated table #655025: 1556181 keys, 69315779 bytes
[default] [JOB 41] Generated table #655026: 797409 keys, 35651472 bytes
[default] [JOB 41] Generated table #655027: 1612608 keys, 69391908 bytes
[default] [JOB 41] Generated table #655028: 462217 keys, 19957191 bytes
[default] [JOB 41] Compacted 4@0 + 4@1 files to L1 => 263627170 bytes



If we look at the working instance, we will see how this leading write log is updated, as well as how individual SSTable tables are written, read and merged.


The log and the latest SSTable tables occupy the lion's share of I / O operations.

If you look at the SSTable statistics on a production server, you will see four “levels” of files, with increasing file sizes at each higher level.

** Compaction Stats [default] **
Level    Files   Size(MB} Score Read(GB}  Rn(GB} Rnp1(GB} Write(GB} Wnew(GB} Moved(GB} W-Amp
--------------------------------------------------------------------------------------------
  L0      1/0      14.46   0.2      0.0     0.0      0.0       0.1      0.1       0.0   0.0
  L1      4/0     194.95   0.8      0.5     0.1      0.4       0.5      0.1       0.0   4.7
  L2     48/0    2551.71   1.0      1.4     0.1      1.3       1.4      0.1       0.0  10.7
  L3    351/0   21735.77   0.8      2.0     0.1      1.9       1.9     -0.0       0.0  14.3
 Sum    404/0   24496.89   0.0      3.9     0.4      3.5       3.9      0.3       0.0  34.2
 Int      0/0       0.00   0.0      3.9     0.4      3.5       3.9      0.3       0.0  34.2

Rd(MB/s} Wr(MB/s} Comp(sec} Comp(cnt} Avg(sec} KeyIn KeyDrop
    0.0     15.6         7         8    0.925       0      0
   20.9     20.8        26         2   12.764     12M     40
   19.4     19.4        73         2   36.524     34M     14
   18.1     16.9       112         2   56.138     52M  3378K
   18.2     18.1       218        14   15.589     98M  3378K
   18.2     18.1       218        14   15.589     98M  3378K

RocksDB stores indexes and Bloom filters of specific SSTable tables in these tables themselves - and they are loaded into memory. These filters and indexes are then polled to find a specific key, and then the full SSTable table is loaded into memory as part of the LRU.

In the vast majority of cases, we see new messages that make our classic deduplication system a classic case of using Bloom filters.

Bloom filters say whether the key is "probably in the plural" or "definitely not in the plural." To give an answer, the filter saves a lot of bits after applying different hash functions for each element that occurred earlier. If all the bits from the hash function converge with the set, then it gives the answer “probably belongs to the set”.


Query the letter w in the Bloom filter when our set contains only {x, y, z}. The filter returns the answer “does not belong to the set”, since one of the bits does not converge.

If the answer is “probably belongs to the set”, then RocksDB can request the initial data from our SSTable tables and determine whether the element is actually present in the set. But in most cases, we can avoid querying any tables at all, because the filter returns the answer “definitely does not belong to the set”.

When we turn to RocksDB, we create a MultiGet request for all relevantmessageIdthat we want to request. We create it as part of a package for performance and to avoid many concurrent blocking operations. It also allows us to pack data from Kafka, and usually avoids random writes in favor of sequential ones.

This explains how read / write tasks show high performance - but there still remains the question of how stale data is considered obsolete.

Delete: snap to size, not time

In our deduplication process, you need to decide whether to limit the system to a strict “deduplication window” or to the total size of the database on disk.

To avoid a system crash due to excessive deduplication for all users, we decided to choose a size limit rather than a time interval limit . This allows you to set a maximum size for each RocksDB instance and deal with sudden jumps or increased load. A side effect is that the time interval can be reduced to less than 24 hours, and at this border our engineer on duty is called.

We periodically declare old keys from RocksDB obsolete to prevent it from growing to an unlimited size. For this we storesecondary index of keys by sequence number, so that we can delete the oldest keys first.

Instead of using RocksDB TTL, which would require maintaining a fixed TTL when opening the database, we delete the objects themselves by the serial number of each nested key.

Since serial numbers are stored as a secondary index, we can quickly request them and “mark” them as deleted. Here is our delete function after passing the serial number to it.

func (d *DB) delete(n int) error {
        // open a connection to RocksDB
        ro := rocksdb.NewDefaultReadOptions()
        defer ro.Destroy()
        // find our offset to seek through for writing deletes
        hint, err := d.GetBytes(ro, []byte("seek_hint"))
        if err != nil {
                return err
        }
        it := d.NewIteratorCF(ro, d.seq)
        defer it.Close()
        // seek to the first key, this is a small
        // optimization to ensure we don't use `.SeekToFirst()`
        // since it has to skip through a lot of tombstones.
        if len(hint) > 0 {
                it.Seek(hint)
        } else {
                it.SeekToFirst()
        }
        seqs := make([][]byte, 0, n)
        keys := make([][]byte, 0, n)
        // look through our sequence numbers, counting up
        // append any data keys that we find to our set to be
        // deleted
        for it.Valid() && len(seqs) < n {
                k, v := it.Key(), it.Value()
                key := make([]byte, len(k.Data()))
                val := make([]byte, len(v.Data()))
                copy(key, k.Data())
                copy(val, v.Data())
                seqs = append(seqs, key)
                keys = append(keys, val)
                it.Next()
                k.Free()
                v.Free()
        }
        wb := rocksdb.NewWriteBatch()
        wo := rocksdb.NewDefaultWriteOptions()
        defer wb.Destroy()
        defer wo.Destroy()
        // preserve next sequence to be deleted.
        // this is an optimization so we can use `.Seek()`
        // instead of letting `.SeekToFirst()` skip through lots of tombstones.
        if len(seqs) > 0 {
                hint, err := strconv.ParseUint(string(seqs[len(seqs)-1]), 10, 64)
                if err != nil {
                        return err
                }
                buf := []byte(strconv.FormatUint(hint+1, 10))
                wb.Put([]byte("seek_hint"), buf)
        }
        // we not only purge the keys, but the sequence numbers as well
        for i := range seqs {
                wb.DeleteCF(d.seq, seqs[i])
                wb.Delete(keys[i])
        }
        // finally, we persist the deletions to our database
        err = d.Write(wo, wb)
        if err != nil {
                return err
        }
        return it.Err()
}

To further guarantee high write speeds, RocksDB does not return immediately and does not delete the key (you remember that SSTable tables are immutable!). Instead, RocksDB adds a “gravestone” to the key, which is then removed during the base compaction process. Therefore, we can quickly make records obsolete during sequential write operations and avoid clogging memory when deleting old items.

Data Correctness

We have already discussed how speed, scaling and cheap search in billions of messages are provided. The last fragment remains - how to ensure the correctness of data in case of various failures.

EBS Snapshots and Applications

To protect our RocksDB instances from damage due to a programmer error or EBS malfunction, we periodically take snapshots of each of our hard drives. Although EBS is replicated on its own, this measure protects against damage caused by some internal mechanism. If we need a specific instance, the client can be paused, and at this time the corresponding EBS disk is unmounted, and then reattached to the new instance. As long as we keep the partition ID unchanged, re-attaching the drive remains a completely painless procedure that still ensures that the data is correct.

In the event of a worker failure, we rely on the write-ahead log of RocksDBso as not to lose messages. Messages are not allowed from the input section until we have a guarantee that RocksDB reliably saved the message in the log.

Read output section

You may have noticed that up to this point there has not been that “atomic” step that ensures that messages are delivered strictly once. At any time, there is a chance that our worker will crash: when writing to RocksDB, when publishing to the output section, or when confirming incoming messages.

We need an atomic “fixation” point that will uniquely cover transactions for all these separate systems. It takes some kind of “source of truth” for our data.

This is where reading from the output section comes into play.

If for some reason the worker crashes or an error occurs in Kafka and then restarts, the first thing is to check with the "source of truth" about whether the event occurred: this source is the output section .

If the message is found in the output section, but not in RocksDB (and vice versa), then the deduplication worker will do the necessary editing to synchronize the database and RocksDB. Essentially, we use the output partition at the same time as the write-ahead log and the ultimate source of truth, while RocksDB captures and validates it.

In real work

Now our deduplication system has been operating in production for three months now, and we are incredibly pleased with the results. If in numbers, then we have:

• 1.5 TB of keys stored on disk in RocksDB
• 4-week deduplication window before deprecating old keys
• approximately 60 billion keys stored in RocksDB instances
• 200 billion messages go through a deduplication system

The system as a whole is fast, efficient and fault tolerant - and with a very simple architecture.

In particular, the second version of our system has a number of advantages over the old deduplication system.

Previously, we stored all the keys in Memcached and used the atomic operator for checking and setting the value of the CAS (check-and-set) record to set non-existent keys. Memcached acted as the fixation point and “atomicity” in the publication of keys.

Although such a scheme worked pretty well, it required a large amount of memory to fit all of our keys. Moreover, we had to choose between accepting random Memcached crashes or doubling our cost of creating memory-hungry fail-safe copies.

The Kafka / RocksDB scheme provides almost all the advantages of the old system, but with increased reliability. Summing up, here are the main achievements:

Storage of data on disk: saving the entire set of keys or full indexing in memory was unacceptably expensive. By transferring more data to disk and using different levels of files and indexes, we were able to significantly reduce the cost. Now we can switch to “cold” storage (EBS) upon failure, and not support the operation of additional “hot” instances in case of failure.

Partitioning:Of course, in order to narrow our search space and avoid loading too many indexes into memory, a guarantee is required that certain messages are sent to the correct workers. Partitioning in Kafka allows you to consistently route these messages along the correct routes, so that we can cache data and generate requests much more efficiently.

Accurate recognition of obsolete keys : in Memcached, we would set a TTL for each key to determine its lifetime, and then rely on the Memcached process to exclude the keys. In the case of large data packets, this would threaten a lack of memory and jumps in CPU usage due to a large number of key exceptions. By instructing the client to delete the keys, we can gracefully avoid the problem by reducing the “deduplication window”.

Kafka as a source of truth : in order to truly avoid deduplication with multiple fixation points, we need to use a truth source that is common to all our customers downstream. Kafka, as such a “source of truth,” works surprisingly well. In the case of most failures (except for failures of Kafka itself) messages are either written to Kafka or not recorded. And using Kafka ensures that published messages are delivered properly and replicated across disks on multiple machines, without having to store large amounts of data in memory.

Batch reading and writing:By doing batch I / O operations for calls to Kafka and RocksDB, we were able to greatly improve performance using sequential read and write. Instead of random access, which was earlier with Memcached, we achieved much higher throughput by improving disk performance and storing only indexes in memory.

In general, we are quite satisfied with the guarantees that the deduplication system we created provides. Using Kafka and RocksDB as the basis for streaming applications is becoming more and more the norm . And we will gladly continue the development of new distributed applications on this foundation.