Scene trending #327

dholms · 2022-11-08T22:34:48Z

This adds a message queue abstraction to the database for handling side effects of created records.

It is currently used for

notifications
scene processing

For scene processing, we maintain two tables: scene_member_count & scene_votes_on_post. And we publish a new trending record when a post has atleast 2 upvotes from a scene & the ratio of upvotes is >=20% of the scene

pfrazee

Yeah this is looking great. Don't let my comments come off as negativity, this is feeling very clean and getting solid value for the amount of effort.

packages/pds/src/db/message-queue/index.ts

pfrazee · 2022-11-10T03:42:22Z

packages/pds/src/db/message-queue/tables/messageQueue.ts

+export interface MessageQueue {
+  id: Generated<number>
+  message: string
+  read: 0 | 1


This is small and completely fine w/you ignoring it, but a separate "MessageQueueCursors" table might be better for two reasons -- it allows multiple separate processors (may not need it but who knows) and it would eliminate N numbers for N events, saving ... some ... space on disk

this should still allow for multiple separate processors. we lock rows with SELECT FOR UPDATE that keep separate processors from stepping on each other's toes

I'm thinking 2 processors that aren't coordinated with each other, eg maintaining their own cursors. But it's not a real concern tbh

packages/pds/src/db/message-queue/messages.ts

packages/pds/src/db/message-queue/index.ts

dholms · 2022-11-11T03:27:00Z

packages/pds/tests/seeds/basic.ts

@@ -42,6 +84,9 @@ export default async (sc: SeedClient) => {
  await sc.vote('down', carol, sc.posts[alice][1].ref)
  await sc.vote('up', carol, sc.posts[alice][2].ref)
  await sc.vote('up', dan, sc.posts[alice][1].ref)
+  await sc.vote('up', alice, sc.posts[carol][0].ref)
+  await sc.vote('up', bob, sc.posts[carol][0].ref)
+  mq && (await mq.process())


here, we want to wait till the queue is done processing before moving on

this gives us a predictable order for our snapshots

devinivy

Had a few thoughts, but so many things I'm digging about this. Real impressive

Approving since I think it's essentially all set, but this comment could probably use attention: https://github.com/bluesky-social/atproto/pull/327/files/908c1e829a64f6a9e7372d4aed8da6eab22b1c6a#diff-4c1f0ad0cea9ddfe9a8ab78478a07d53a598be698252db06be359db1c77eda21

packages/pds/src/db/index.ts

packages/pds/src/db/migrations/20221021T162202001Z-init.ts

packages/pds/src/db/records/repost.ts

devinivy · 2022-11-11T16:55:58Z

packages/pds/src/index.ts

+  const messageQueue = new SqlMessageQueue('pds', db, (did: string) => {
+    return auth.verifier.loadAuthStore(keypair, [], did)
+  })
+  db.setMessageQueue(messageQueue)


I know you weren't pumped about this. The two ideas that come to mind are that

perhaps indexRecord(), deleteRecord() don't need to live on db.

perhaps the message queue deserves a separate db of its own.

yeah both of these crossed my mind as well 🤔

we may want something like RecordIndexer. This kinda has to do with that fact that the db has a lot of functionality jammed into it. probably needs to be rethought & broken up a bit

devinivy · 2022-11-11T17:03:54Z

packages/pds/src/db/message-queue/index.ts

+        if (this.db.dialect !== 'sqlite') {
+          builder = builder.forUpdate()
+        }


You could add a skipLocked() here so that if there's contention for the cursor, one consumer gets it and the other just bails rather than wait for it to open up.

i think we actually want to wait for it to open up. we basically ping the consumer on every event push to try & process exactly 1 event. So it should wait until the cursor is freed up & then give it a go

Ahh, okay got it. How does keepGoing play into that?

ah not sure if you saw, but keepGoing actually got ripped out.

Yeah beforehand, i was throwing way to many handlers at this & they'd all just hang out at the lock with their hands open until the DB got back to them & said "sorry buddy no cursor for you"

Ahhh right on!

devinivy · 2022-11-11T17:05:11Z

packages/pds/src/db/message-queue/index.ts

+        await this.handleMessage(dbTxn, message)
+        await dbTxn.db
+          .updateTable('message_queue_cursor')
+          .set({ cursor: sql`cursor + 1` })


Only question I have about this is if we can assume the incrementing id is guaranteed to not leave any gaps. Otherwise I think the cursor can get stuck.

oh that's a good point. i think so? worth a look 🧐

it doesn't guarantee it :notlikethis:

kk this should fix it
e446481

(also added an actual SQL LIMIT in the next commit)

devinivy · 2022-11-11T17:11:29Z

packages/pds/src/db/message-queue/index.ts

+  private async handleMessage(db: Database, message: Message) {
+    switch (message.type) {
+      case 'add_member':
+        return this.handleAddMember(db, message)
+      case 'remove_member':
+        return this.handleRemoveMember(db, message)
+      case 'add_upvote':
+        return this.handleAddUpvote(db, message)
+      case 'remove_upvote':
+        return this.handleRemoveUpvote(db, message)
+      case 'create_notification':
+        return this.handleCreateNotification(db, message)
+      case 'delete_notifications':
+        return this.handleDeleteNotifications(db, message)
+    }
+  }


Eventually I think it might be nice to colocate these handlers with related code (e.g.handleAddMember() living closer to member methods) and allow other areas of the app to hook-in to the message queue. Thinking of the message queue as more of a pull-based transport, rather than a vertical slice of the app. Zero problem with it as-is today, though, just thinkin!

yup makes sense 👍

devinivy · 2022-11-11T17:36:02Z

packages/pds/src/db/message-queue/index.ts

+      await this.handleMessage(dbTxn, message)
+      await dbTxn.db
+        .updateTable('message_queue_cursor')
+        .set({ cursor: sql`cursor + 1` })


I think by incrementing by one here we can cause messages to get processed multiple times if there's a gap in the ids. Pretty sure we'll want to catch the cursor up to whichever message was just processed (plus 1).

a true hero you are 🙏

devinivy

💯

* wip * views * trending schema * starting message queue * scene processor * wip * send mq messages from db * db events * undo screwing up codegen lol * setup queue * db migrations * fixing up message processing * div by 0 check * tx issue * queue use cursor * update not insert * sql bugfix + tests * trying to linearize tests * correclty serializing txs * attempt update before insert * log errors * handle gaps in cursor * cleanup * oops reenable test * correctly incr cursor

dholms added 3 commits November 9, 2022 14:45

wip

627402c

views

7cc7cdf

trending schema

0de61ed

dholms force-pushed the scene-trending branch from a2eb100 to 0de61ed Compare November 9, 2022 20:48

dholms added 8 commits November 9, 2022 16:02

starting message queue

4f0d51c

scene processor

1a6cfbb

wip

b07f35a

send mq messages from db

27c6aaf

db events

2d9fd35

undo screwing up codegen lol

ada7b60

setup queue

196b277

db migrations

49d1b14

pfrazee reviewed Nov 10, 2022

View reviewed changes

dholms added 8 commits November 10, 2022 11:33

fixing up message processing

b814d5a

div by 0 check

ce5e20e

tx issue

7d596c4

queue use cursor

12f47f4

update not insert

fa48289

emrge

74c4787

sql bugfix + tests

cca2c3d

trying to linearize tests

908c1e8

dholms commented Nov 11, 2022

View reviewed changes

dholms marked this pull request as ready for review November 11, 2022 03:27

dholms added 3 commits November 11, 2022 10:59

correclty serializing txs

8b62ea5

attempt update before insert

b6345b7

log errors

6771938

devinivy approved these changes Nov 11, 2022

View reviewed changes

dholms added 2 commits November 11, 2022 11:28

handle gaps in cursor

e446481

cleanup

7fa5203

devinivy reviewed Nov 11, 2022

View reviewed changes

dholms added 2 commits November 11, 2022 12:17

oops reenable test

e7cdd32

correctly incr cursor

1a7d10c

devinivy approved these changes Nov 11, 2022

View reviewed changes

dholms merged commit 962041e into main Nov 11, 2022

dholms deleted the scene-trending branch November 11, 2022 18:29

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Scene trending #327

Scene trending #327

dholms commented Nov 8, 2022 •

edited

Loading

pfrazee left a comment

pfrazee Nov 10, 2022

dholms Nov 10, 2022

pfrazee Nov 10, 2022

dholms Nov 11, 2022

devinivy left a comment

devinivy Nov 11, 2022

dholms Nov 11, 2022

devinivy Nov 11, 2022

dholms Nov 11, 2022

devinivy Nov 11, 2022

dholms Nov 11, 2022

devinivy Nov 11, 2022

devinivy Nov 11, 2022

dholms Nov 11, 2022

dholms Nov 11, 2022

dholms Nov 11, 2022

devinivy Nov 11, 2022

dholms Nov 11, 2022

devinivy Nov 11, 2022

dholms Nov 11, 2022

devinivy left a comment

Scene trending #327

Scene trending #327

Conversation

dholms commented Nov 8, 2022 • edited Loading

pfrazee left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

devinivy left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

devinivy left a comment

Choose a reason for hiding this comment

dholms commented Nov 8, 2022 •

edited

Loading