Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Sync and Message Storage

Matrix has an endpoint /sync, and other related endpoints, which allows users to retrieve a stream of events. This needs to be considered when implementing event storage if we want to support it at a scale.

Requirements for Message Storage and Retrieval

  • We want to let clients paginate events and fetch them in real time, using the following endpoints:
  • We want to do relational-style queries on events, to find the current state for instance, or find rooms the user has been invited to
  • We want to be able to fetch old events quickly
  • For unencrypted rooms, we want to support server-side full-text search
  • We want to be able to insert events at arbitrary points in the timeline, if the federation has been partitioned for a while
  • We want to be able to support multiple events together
    • When creating a new room, we want to ensure that all the initial events are considered as a bundle. We don’t want a partial room creation.

Sync’s Batch Token

Batch token is a string token that includes the information about the state event streams (Both PDUs and EDUs) in the server. This token is given to the user for different purposes.

In the legacy /sync it is used to fill the since query parameter. This string is not created by the client, rather it is created by the server from previous /sync calls as the next_batch field in the body of the response.

For us this Token looks like this s_123571293 where the first part s_ specifies that it is a sync token, and the other parts are specifying positions in the different streams of events so that the server know until which event the client has received events from that stream.

So for now, the current structure is s_<stream_id>, where the stream_id is specifying the position of the stream for room events (or PDUs).

Roomserver Solution

The roomserver solution would involve, for each sync request, the client service making a request to each room service, such that all rooms were queried for updates. This, however, would make syncing very expensive. Since roomservers are in charge of reading and writing to rooms, sync requests would block writing to a room. The alternative, where the client API makes direct SQL queries to the database, would block far less. In larger deployments, there is also the option of having a secondary, read-only database for the sync API.

Naive Solution

The naive solution for this would be to use the primary key for the events table, which we know is given by a postgresql sequence.

But this has the issue that SQL sequences can end up committed out of order.

Essentially, as transaction A inserts an event, it reserves event id 1. A second writer will start transaction B, insert an event, and reserve event 2. But it is entirely possible that transaction B will complete before transaction A, resulting in a database where there is an event 2, but no event 1. If we then implement sync by paginating through event ids, we will return event id 2, and the next page should logically start at event id 2. But doing this, the event with id 1 will never end up in the stream.

Current Solution

TL;DR

We use an advisory lock to restrict our reads to sequence numbers that have been committed. In crates/libs/db/arminius-lib-db/migrations/00_locking_sequence.sql we define two functions, locking_nextval() and max_safe_seq(), which can be used for this purpose.

When fetching events from the database for sync responses, we simply include a WHERE id <= max_safe_seq('eventids') in order to make sure we don’t advance our index past the point where all sequence numbers belong to committed rows.

The relevant part of the since token is then just the highest stream_id returned.

By only restricting the reads, instead of restricting the writes we avoid throughput being throttled. But a long-running transaction will also hinder all syncs in advancing. The main goal of implementing sync using advisory locks was getting a solution rapidly, which is guaranteed to work, and enabled us to begin implementing the rest of the homeserver. However, this solution has not been tested at all with large server loads, and we therefore expect it to fail. We assume that this decision will have to be revisited at a later point, when we are able to test our homeserver at high load.

On Insertion of Events:

We have a table called events.

On each row, there is a column called “stream_id,” which is a unique 64-bit signed integer.

Filling of this field happens using a locked incrementing sequence (called eventids) by calling the locking_nextval function. This function is called by default inside PostgreSQL.

On the Reading of Events:

This solution is (almost) irrelevant when there is no since token and the full state is required. So we just ignore that part.

So there are two tokens that are important to us

  • The since token that the clients send in the request.
  • The next_batch token that we should give the user in the response.

Therefore, simply put, the server asks PostgreSQL to give us every event with a stream_id greater than the event stream_id (that is parsed from the since token) and smaller than or equivalent to the max safe sequence number.

How the max safe sequence is decided is based on an SQL function we have already defined called max_safe_seq.

max_safe_seq looks into every lock for this sequence, finds the minimum number locked in the sequence, and returns that, or if there is no lock, it returns the maximum 64-bit signed integer (9_223_372_036_854_775_807).

The next_batch token is the maximum stream_id of the events returned.

The Expected Behavior

Here is the entire timeline I will use to explain the characteristics of the system.

I will continue with a detailed explanation of each part.

sequenceDiagram
    autonumber
    actor c1 as Client 1
    actor c2 as Client 2
    participant sv as Server
    participant pq as Postgresql
    Note over pq: max_safe_seq=9_223_372_036_854_775_807

    %% initial message
    rect AliceBlue
        c1->>+sv: send_message
        sv-->>+pq: insert event1
        pq-->>-sv: (stream_id: 9)
        sv->>-c1: 200
    end

    %% sync attempt
    rect DarkKhaki
        c2-->>+sv: /sync
        Note over pq,sv: because there is no since=?  we request everything.
        sv-->>+pq: Select every event that is from 0 to the last id that is not in a transaction but is less than everything else that is in it.
        Note over pq: max_safe_seq=9_223_372_036_854_775_807
        pq-->>-sv: [event1]
        sv -->>-c2: [event1], next_batch: s_9
    end

    %% let's do a long transaction...
    rect DarkSeaGreen
        Note over c1,pq: now lets do a long running transaction.
        c1->>+sv: create_room

        sv-->>+pq: BEGIN
        sv-->>+pq: insert event2
        pq-->>-sv: (stream_id: 10)

        %% sync attempt
        rect DarkKhaki
            c2-->>+sv: /sync?=s_9
            sv-->>+pq: Select every event that is from 9 to the last id that is not in a transaction but is less than everything else that is in it.
            Note over pq: max_safe_seq=9
            pq-->>-sv: [<empty>]
            sv -->>-c2: [<empty>], next_batch: s_9
        end

        sv-->>+pq: insert event3
        pq-->>-sv: (stream_id: 11)

        %%  this is to show that this message is not going to be shown
        rect AliceBlue
            Note over c2,sv: client 2 sends a message
            c2->>+sv: send_message
            sv-->>+pq: insert event4
            pq-->>-sv: (stream_id: 12)
            sv->>-c2: 200
        end

        %% sync attempt
        rect DarkKhaki
            c2-->>+sv: /sync?=s_9
            sv-->>+pq: Select every event that is from 9 to the last id that is not in a transaction but is less than everything else that is in it.
            Note over pq: max_safe_seq=9
            Note over c2,sv: but it will not receive anything yet.
            Note over c2,sv: because the stream_id is more than 9 which is the current max_safe_seq.
            pq-->>-sv: [<empty>]
            sv -->>-c2: [<empty>], next_batch: s_9
        end

        sv-->>+pq: insert event5
        pq-->>-sv: (stream_id: 13)

        sv-->>+pq: COMMIT
        pq->>-sv: ok
        sv->>-c1: 200
    end

    %% sync attempt
    rect DarkKhaki
        c2-->>+sv: /sync?=s_9
        sv-->>+pq: Select every event that is from 9 to the last id that is not in a transaction but is less than everything else that is in it.
        Note over pq: max_safe_seq=9_223_372_036_854_775_807
        pq-->>-sv: [event2, event3, event4, event5]
        sv -->>-c2: [event2, event3, event4, event5], next_batch: s_13
    end

1. Normal Insertion of an Event

Let’s imagine that we evaluate the max_safe_seq() function when there is no transaction running; the result would be the maximum 64-bit signed integer.

  1. Initially, the highest stream_id in the database is 8, and no stream_ids are locked.
  2. To insert a message. Client 1 sends a request to the server.
  3. The server implicitly starts a transaction by sending an insert query to the database. While this is going on, the max_safe_seq() will evaluate to 8.
  4. PostgreSQL returns success.
  5. and the server returns success to the client.
sequenceDiagram
    autonumber
    actor c1 as Client 1
    actor c2 as Client 2
    participant sv as Server
    participant pq as Postgresql
    Note over pq: max_safe_seq=9_223_372_036_854_775_807

    %% initial message
    rect AliceBlue
        c1->>+sv: `send_message`
        sv-->>+pq: insert event1
        pq-->>-sv: (stream_id: 9)
        sv->>-c1: 200
    end

Normal business as usual.

2. First Sync

Client 2 now wants to sync.

  1. It sends a sync request without any since tokens. For the sake of a story, imagine this is their first sync with the new device! Yaaaay!
  2. The server requests every event that has a stream_id less than max_safe_seq. Here, because there is no transaction running, the max_safe_seq evaluates to i64::MAX, which basically covers every event.
  3. The only event in our imaginary awesome introverted server is event1, which was the message sent by client 1 in the previous section.
  4. server returns that event to the client, along with a token with the stream_id of the event1, which is 9. (The token in this example is s_9)
sequenceDiagram
    actor c1 as Client 1
    actor c2 as Client 2
    participant sv as Server
    participant pq as Postgresql
    Note over pq: max_safe_seq=9_223_372_036_854_775_807
    %% sync attempt
    rect DarkKhaki
        c2-->>+sv: `/sync`
        Note over pq,sv: because there is no `since=?`  we request everything.
        sv-->>+pq: Select every event that is from 0 to the last id that is not in a transaction but is less than everything else that is in it.
        Note over pq: max_safe_seq=9_223_372_036_854_775_807
        pq-->>-sv: [event1]
        sv -->>-c2: [event1], next_batch: s_9
    end

3. The Long Transaction

Now let’s do something more ambitious.

Let’s see what happens when there is a longer transaction going on afterward.

Let’s say that client 1 is creating a room. And client 2, being the socially active person we know and love, is syncing and sending messages to another while client 1 is at it.

sequenceDiagram
    autonumber
    actor c1 as Client 1
    actor c2 as Client 2
    participant sv as Server
    participant pq as Postgresql
    Note over pq: max_safe_seq=9_223_372_036_854_775_807
    %% let's do a long transaction...
    rect DarkSeaGreen
        Note over c1,pq: now lets do a long running transaction.
        c1->>+sv: `create_room`

        sv-->>+pq: BEGIN
        sv-->>+pq: insert event2
        pq-->>-sv: (stream_id: 10)

        %% sync attempt
        rect DarkKhaki
            c2-->>+sv: `/sync?=s_9`
            sv-->>+pq: Select every event that is from 9 to the last id that is not in a transaction but is less than everything else that is in it.
            Note over pq: max_safe_seq=9
            pq-->>-sv: [<empty>]
            sv -->>-c2: [<empty>], next_batch: s_9
        end

        sv-->>+pq: insert event3
        pq-->>-sv: (stream_id: 11)

        %%  this is to show that this message is not going to be shown
        rect AliceBlue
            Note over c2,sv: client 2 sends a message
            c2->>+sv: `send_message`
            sv-->>+pq: insert event4
            pq-->>-sv: (stream_id: 12)
            sv->>-c2: 200
        end

        %% sync attempt
        rect DarkKhaki
            c2-->>+sv: `/sync?=s_9`
            sv-->>+pq: Select every event that is from 9 to the last id that is not in a transaction but is less than everything else that is in it.
            Note over pq: max_safe_seq=9
            Note over c2,sv: but it will not receive anything yet.
            Note over c2,sv: because the stream_id is more than 9 which is the current max_safe_seq.
            pq-->>-sv: [<empty>]
            sv -->>-c2: [<empty>], next_batch: s_9
        end

        sv-->>+pq: insert event5
        pq-->>-sv: (stream_id: 13)

        sv-->>+pq: COMMIT
        pq->>-sv: ok
        sv->>-c1: 200
    end
  1. Client 1 sends a create_room request to the server.
  2. Server starts a transaction.
  3. Server inserts a transaction.
  4. The stream_id is at 10 now.
  5. Client 2 sends a sync request to the server with the s_9 token they received from the previous section.
  6. The server queries PostgreSQL for every event that has a stream_id more than 9 and less than or equal to the max_safe_seq at this point, which is also 9. So, basically nothing.
  7. No rows are returned.
  8. The next_batch token here will be the same token as the since token.
  9. The server, continuing the request of client 1, inserts event3…
  10. database sends back stream_id of 11.
  11. Client 2, excited, now sends a message.
  12. event4 is the client2 send_message event now inserted into the database…
  13. …with stream_id of 12
  14. 200 to the client 2 for their message request is returned.
  15. Client 2 feels insecure, so they sync once more; maybe client 1 has already responded. Poor guy.
  16. Again, the max_safe_seq is still at 9 because the transaction has not released the lock yet, so…
  17. Nothing is returned.
  18. and the next_batch token will remain the same. They don’t even see their messaging event, adding to the feeling of ghosting.
  19. server sends a query for inserting event5.
  20. with stream_id = 13.
  21. and the server sends COMMIT.
  22. PostgreSQL sends back success for the transaction commit.
  23. Client 1, unaware of the long, gigantic drama that awaits them, receives a 200 for the room creation request.

4. The Final Sync

Now for the final sync.

Client 2 sends another sync request with the since token of s_9.

This time the max_safe_seq evaluates to the i64::MAX (because there is no other transaction going that locks the sequence).

This means that event2, event3, event4 and event5 are all available.

Client 2, facing the reality of what they have done, will receive all these events with an eventually changed next_batch token of s_13.

sequenceDiagram

    actor c1 as Client 1
    actor c2 as Client 2
    participant sv as Server
    participant pq as Postgresql
    Note over pq: max_safe_seq=9_223_372_036_854_775_807

    %% sync attempt
    rect DarkKhaki
        c2-->>+sv: `/sync?=s_9`
        sv-->>+pq: Select every event that is from 9 to the last id that is not in a transaction but is less than everything else that is in it.
        Note over pq: max_safe_seq=9_223_372_036_854_775_807
        pq-->>-sv: [event2, event3, event4, event5]
        sv -->>-c2: [event2, event3, event4, event5], next_batch: s_13
    end