Add seqno_creation to be able to skip irrelevant deletions

Add a new column messages.seqno_creation that lets us track a message's
initial seqno value.  This allows us to skip deletion tombstones for
messages that the user doesn't have (i.e. because the message was both
created and then deleted after the client's current seqno) which can
significantly drop the number of messages we have to deliver when
syncing up a room with a lot of deletions.

Author: jagerman

sogs/migrations/__init__.py

@@ -13,6 +13,7 @@
     reactions,
     room_accessible,
     room_moderators,
+    seqno_creation,
     seqno_etc,
     user_permissions,
     user_perm_futures,
@@ -41,6 +42,7 @@ def migrate(conn, *, check_only=False):
         new_columns,
         seqno_etc,
         reactions,
+        seqno_creation,
         message_views,
         user_perm_futures,
         room_accessible,

sogs/migrations/message_views.py

@@ -61,11 +61,11 @@ def migrate(conn, *, check_only):
         conn.execute(
             """
 CREATE VIEW message_metadata AS
-SELECT id, room, "user", session_id, posted, edited, seqno, seqno_data, seqno_reactions,
+SELECT id, room, "user", session_id, posted, edited, seqno, seqno_data, seqno_reactions, seqno_creation,
         filtered, whisper_to, whisper_mods,
         length(data) AS data_unpadded, data_size, length(signature) as signature_length
     FROM message_details
-"""
+"""  # noqa: E501
         )
 
     else:  # postgresql
@@ -100,11 +100,11 @@ def migrate(conn, *, check_only):
 -- View of `messages` that is useful for manually inspecting table contents by only returning the
 -- length (rather than raw bytes) for data/signature.
 CREATE VIEW message_metadata AS
-SELECT id, room, "user", session_id, posted, edited, seqno, seqno_data, seqno_reactions,
+SELECT id, room, "user", session_id, posted, edited, seqno, seqno_data, seqno_reactions, seqno_creation,
         filtered, whisper_to, whisper_mods,
         length(data) AS data_unpadded, data_size, length(signature) as signature_length
     FROM message_details;
-            """
+"""  # noqa: E501
         )
 
     return True

sogs/migrations/seqno_creation.py

@@ -0,0 +1,58 @@
+import logging
+from .exc import DatabaseUpgradeRequired
+
+
+def migrate(conn, *, check_only):
+    """
+    Adds a seqno_creation column to track the seqno when a message was created so that we can skip
+    deleted messages entirely (i.e. omit the tombstone) when polling from a seqno before the message
+    was created.
+    """
+
+    from .. import db
+
+    if 'seqno_creation' in db.metadata.tables['messages'].c:
+        return False
+
+    if check_only:
+        raise DatabaseUpgradeRequired("message creation seqno")
+
+    logging.warning("Adding messages.seqno_creation column")
+    if db.engine.name == 'sqlite':
+        conn.execute("ALTER TABLE messages ADD COLUMN seqno_creation INTEGER NOT NULL DEFAULT 0")
+        conn.execute("DROP TRIGGER IF EXISTS messages_insert_counter")
+        conn.execute(
+            """
+CREATE TRIGGER messages_insert_counter AFTER INSERT ON messages
+FOR EACH ROW
+BEGIN
+    UPDATE rooms SET message_sequence = message_sequence + 1 WHERE id = NEW.room;
+    UPDATE messages SET seqno_data = (SELECT message_sequence FROM rooms WHERE id = NEW.room) WHERE id = NEW.id;
+    UPDATE messages SET seqno_creation = seqno_data WHERE id = NEW.id;
+END
+"""  # noqa: E501
+        )
+    else:  # postgresql
+        conn.execute(
+            """
+ALTER TABLE messages ADD COLUMN seqno_creation BIGINT NOT NULL DEFAULT 0;
+
+CREATE OR REPLACE FUNCTION trigger_messages_insert_counter()
+RETURNS TRIGGER LANGUAGE PLPGSQL AS $$
+DECLARE
+    new_seqno BIGINT := increment_room_sequence(NEW.room);
+BEGIN
+    UPDATE messages SET seqno_data = new_seqno, seqno_creation = new_seqno WHERE id = NEW.id;
+    RETURN NULL;
+END;$$;
+DROP TRIGGER IF EXISTS messages_insert_counter ON messages;
+CREATE TRIGGER messages_insert_counter AFTER INSERT ON messages
+FOR EACH ROW EXECUTE PROCEDURE trigger_messages_insert_counter();
+"""
+        )
+
+    # Drop these to be recreated (with the no column) in the message_views migration.
+    conn.execute("DROP VIEW IF EXISTS message_metadata")
+    conn.execute("DROP VIEW IF EXISTS message_details")
+
+    return True

sogs/model/room.py

@@ -600,8 +600,15 @@ def get_messages_for(
             if after <= max_old_id:
                 after += offset
 
-        # We include deletions only if doing a sequence update request:
-        not_deleted_clause = '' if sequence is not None else 'AND data IS NOT NULL'
+        # We include deletions only if doing a sequence update request, but only include deletions
+        # for messages that were created *before* the given sequence number (the client won't have
+        # messages created after that, so it is pointless to send them tombstones for messages they
+        # don't know about).
+        not_deleted_clause = (
+            'AND (data IS NOT NULL OR seqno_creation <= :sequence)'
+            if sequence is not None
+            else 'AND data IS NOT NULL'
+        )
         message_clause = (
             'AND seqno > :sequence AND seqno_data > :sequence'
             if sequence is not None and not reaction_updates

sogs/schema.pgsql

@@ -39,6 +39,7 @@ CREATE TABLE messages (
     seqno BIGINT NOT NULL DEFAULT 0, /* set to the room's `message_sequence` counter when any of the individual seqno values are updated */
     seqno_data BIGINT NOT NULL DEFAULT 0, /* updated when `data` changes (i.e. edits, deletions) */
     seqno_reactions BIGINT NOT NULL DEFAULT 0, /* updated when reactions are added/removed */
+    seqno_creation BIGINT NOT NULL DEFAULT 0, /* set to the seqno at the time of creation (and not updated afterwards) */
     data BYTEA, /* Actual message content, not including trailing padding; set to null to delete a message */
     data_size BIGINT, /* The message size, including trailing padding (needed because the signature is over the padded data) */
     signature BYTEA, /* Signature of `data` by `public_key`; set to null when deleting a message */
@@ -72,8 +73,11 @@ END;$$;
 -- Trigger to increment a room's `message_sequence` counter and assign it to the message's `seqno`
 -- field for new messages.
 CREATE OR REPLACE FUNCTION trigger_messages_insert_counter()
-RETURNS TRIGGER LANGUAGE PLPGSQL AS $$BEGIN
-    UPDATE messages SET seqno_data = increment_room_sequence(NEW.room) WHERE id = NEW.id;
+RETURNS TRIGGER LANGUAGE PLPGSQL AS $$
+DECLARE
+    new_seqno BIGINT := increment_room_sequence(NEW.room);
+BEGIN
+    UPDATE messages SET seqno_data = new_seqno, seqno_creation = new_seqno WHERE id = NEW.id;
     RETURN NULL;
 END;$$;
 CREATE TRIGGER messages_insert_counter AFTER INSERT ON messages
@@ -330,7 +334,7 @@ EXECUTE PROCEDURE trigger_message_details_deleter();
 -- View of `messages` that is useful for manually inspecting table contents by only returning the
 -- length (rather than raw bytes) for data/signature.
 CREATE VIEW message_metadata AS
-SELECT id, room, "user", session_id, posted, edited, seqno, seqno_data, seqno_reactions,
+SELECT id, room, "user", session_id, posted, edited, seqno, seqno_data, seqno_reactions, seqno_creation,
         filtered, whisper_to, whisper_mods,
         length(data) AS data_unpadded, data_size, length(signature) as signature_length
     FROM message_details;

sogs/schema.sqlite

@@ -38,6 +38,7 @@ CREATE TABLE messages (
     seqno INTEGER NOT NULL DEFAULT 0, /* set to the room's `message_sequence` counter when any of the individual seqno values are updated */
     seqno_data INTEGER NOT NULL DEFAULT 0, /* updated when `data` changes (i.e. edits, deletions) */
     seqno_reactions INTEGER NOT NULL DEFAULT 0, /* updated when reactions are added/removed */
+    seqno_creation INTEGER NOT NULL DEFAULT 0, /* set to the seqno at the time of creation (and not updated afterwards) */
     data BLOB, /* Actual message content, not including trailing padding; set to null to delete a message */
     data_size INTEGER, /* The message size, including trailing padding (needed because the signature is over the padded data) */
     signature BLOB, /* Signature of `data` by `public_key`; set to null when deleting a message */
@@ -65,6 +66,7 @@ FOR EACH ROW
 BEGIN
     UPDATE rooms SET message_sequence = message_sequence + 1 WHERE id = NEW.room;
     UPDATE messages SET seqno_data = (SELECT message_sequence FROM rooms WHERE id = NEW.room) WHERE id = NEW.id;
+    UPDATE messages SET seqno_creation = seqno_data WHERE id = NEW.id;
 END;
 
 -- Trigger to do various tasks needed when a message is edited/deleted:
@@ -280,7 +282,7 @@ END;
 -- View of `messages` that is useful for manually inspecting table contents by only returning the
 -- length (rather than raw bytes) for data/signature.
 CREATE VIEW message_metadata AS
-SELECT id, room, "user", session_id, posted, edited, seqno, seqno_data, seqno_reactions,
+SELECT id, room, "user", session_id, posted, edited, seqno, seqno_data, seqno_reactions, seqno_creation,
         filtered, whisper_to, whisper_mods,
         length(data) AS data_unpadded, data_size, length(signature) as signature_length
     FROM message_details;

tests/test_room_routes.py

@@ -687,6 +687,85 @@ def test_fetch_since(client, room, user, no_rate_limit):
     assert fetches == (sum(counts) + 24) // 25
 
 
+def test_fetch_since_skip_deletions(client, room, user, no_rate_limit):
+    # Insert 10 posts; they will have seqno == id (i.e. 1 to 10).
+    for i in range(1, 11):
+        room.add_post(user, f"fake data {i}".encode(), pad64(f"fake sig {i}"))
+
+    # Delete some:
+    deleted = (2, 4, 5, 8, 9)
+    for i in deleted:
+        r = sogs_delete(client, f'/room/test-room/message/{i}', user)
+        assert r.status_code == 200
+
+    def get_and_clean_since(seqno):
+        r = sogs_get(client, f"/room/test-room/messages/since/{seqno}", user)
+        assert r.status_code == 200
+        res = r.json
+        for m in res:
+            for k in ('posted', 'session_id', 'reactions'):
+                if k in m:
+                    del m[k]
+            for k in ('data', 'signature', 'edited'):
+                if k in m and m[k] is not None:
+                    m[k] = True
+        return res
+
+    # If we poll from 1 we should only see the messages (skipping the first one with seqno=1) that
+    # remain (since our polling seqno is before the deleted messages were created in the first
+    # place):
+    assert get_and_clean_since(1) == [
+        {'id': i, 'seqno': i, 'data': True, 'signature': True} for i in (3, 6, 7, 10)
+    ]
+
+    def deleted_entry(id, seqno):
+        return {'id': id, 'seqno': seqno, 'edited': True, 'deleted': True, 'data': None}
+
+    # If we poll from 2 we should get the deletion for 2, but not the higher deletions
+    assert get_and_clean_since(2) == [
+        *({'id': i, 'seqno': i, 'data': True, 'signature': True} for i in (3, 6, 7, 10)),
+        *(deleted_entry(i, s) for i, s in ((2, 11),)),
+    ]
+
+    # From 4 we should get deletions 2 and 4
+    assert get_and_clean_since(4) == [
+        *({'id': i, 'seqno': i, 'data': True, 'signature': True} for i in (6, 7, 10)),
+        *(deleted_entry(i, s) for i, s in ((2, 11), (4, 12))),
+    ]
+
+    # and so on
+    assert get_and_clean_since(5) == [
+        *({'id': i, 'seqno': i, 'data': True, 'signature': True} for i in (6, 7, 10)),
+        *(deleted_entry(i, s) for i, s in ((2, 11), (4, 12), (5, 13))),
+    ]
+    assert get_and_clean_since(6) == [
+        *({'id': i, 'seqno': i, 'data': True, 'signature': True} for i in (7, 10)),
+        *(deleted_entry(i, s) for i, s in ((2, 11), (4, 12), (5, 13))),
+    ]
+    assert get_and_clean_since(7) == [
+        *({'id': i, 'seqno': i, 'data': True, 'signature': True} for i in (10,)),
+        *(deleted_entry(i, s) for i, s in ((2, 11), (4, 12), (5, 13))),
+    ]
+
+    assert get_and_clean_since(9) == [
+        *({'id': i, 'seqno': i, 'data': True, 'signature': True} for i in (10,)),
+        *(deleted_entry(i, s) for i, s in ((2, 11), (4, 12), (5, 13), (8, 14), (9, 15))),
+    ]
+    assert get_and_clean_since(10) == [
+        *(deleted_entry(i, s) for i, s in ((2, 11), (4, 12), (5, 13), (8, 14), (9, 15))),
+    ]
+    assert get_and_clean_since(11) == [
+        *(deleted_entry(i, s) for i, s in ((4, 12), (5, 13), (8, 14), (9, 15))),
+    ]
+    assert get_and_clean_since(13) == [
+        *(deleted_entry(i, s) for i, s in ((8, 14), (9, 15))),
+    ]
+    assert get_and_clean_since(14) == [
+        *(deleted_entry(i, s) for i, s in ((9, 15),)),
+    ]
+    assert get_and_clean_since(15) == []
+
+
 def test_fetch_before(client, room, user, no_rate_limit):
     for i in range(1000):
         room.add_post(user, f"data-{i}".encode(), pad64(f"fake sig {i}"))