fix: enable connection revalidation to scale to more connections than can fit in memory at once #491
Conversation
…n can fit in memory at once
|
The latest updates on your projects. Learn more about Vercel for Git ↗︎
|
| } | ||
|
|
||
| const CONNECTION_KEY_PREFIX = 'connection/'; | ||
| const CONNECTION_ROOM_INDEX_PREFIX = 'connection_room/'; |
There was a problem hiding this comment.
I feel like connections_by_room would be a more immediately obvious name.
| return `${CONNECTION_KEY_PREFIX}${encodeURIComponent(userID)}/`; | ||
| } | ||
|
|
||
| function connectionKeyToConnectionRoomIndexString(key: ConnectionKey): string { |
There was a problem hiding this comment.
Do you purposely end it with /?
There was a problem hiding this comment.
i was just keeping this consistent with what I did for the connectionKeyToString and I can't remember why I ended that with slash.
| } | ||
|
|
||
| function connectionKeyToConnectionRoomIndexString(key: ConnectionKey): string { | ||
| return `${CONNECTION_ROOM_INDEX_PREFIX}${encodeURIComponent( |
There was a problem hiding this comment.
nit, but you could use the below function inside this function to ensure they remain in sync.
There was a problem hiding this comment.
done. I also reworked this a bit so the index key reuses the connection key. it makes the index key longer, but will help ensure things keep in sync, and its closer to the approach you would take with a general index utility.
| connectTimestamp: Date.now(), | ||
| }; | ||
| await this._state.storage.put(connectionKey, connectionRecord); | ||
| recordConnection( |
There was a problem hiding this comment.
This method is getting so long, it's worth thinking about how to break it up for readability. Obviously pieces can be broken up into method helpers, but often doing that doesn't help too much unless the inputs and outputs are very clear. Is it possible/useful to factor out a bare connect() function or a set of bare function helpers? I find that makes things much easier to reason about because the inputs/outputs are reduced to just the parameters and return value vs state shared with the class.
There was a problem hiding this comment.
I'll look at this as a follow up. This has gotten to be a beast of a function (though this particular change doesn't add to it).
| return connectionKeys; | ||
| } | ||
|
|
||
| async function* createConnectionsByRoomGenerator( |
There was a problem hiding this comment.
Naming nit: consider generateConnectionsByRoom.
There was a problem hiding this comment.
+1
Generator is an implementation strategy for an iterator. No need to include that in the name.
| while (true) { | ||
| const nextRoomListResult = await storage.list({ | ||
| startAfter: lastKey, | ||
| prefix: CONNECTIONS_BY_ROOM_INDEX_PREFIX, | ||
| limit: 1, | ||
| }); | ||
| if (nextRoomListResult.size === 0) { | ||
| return; | ||
| } | ||
| const firstRoomIndexString: string = [...nextRoomListResult.keys()][0]; |
There was a problem hiding this comment.
This one is safe as next(). list returns a Map and the map iterators have no return methods so there is no cleanup behavior involved.
There was a problem hiding this comment.
Or just iterator over all (one) of them. Then if we wanted to change the limit to 5 or whatever, things would work.
| const connectionsListResult = await storage.list({ | ||
| startAfter: lastKey, | ||
| prefix: CONNECTION_KEY_PREFIX, | ||
| limit: 1000, |
There was a problem hiding this comment.
I think this could use a comment... same with createConnectionsByRoomGenerator
| const connectionRoomIndexString = | ||
| connectionKeyToConnectionRoomIndexString(connectionKey); | ||
| // no await to ensure the two puts are coalesced and done atomically | ||
| void storage.put(connectionKeyString, record); |
There was a problem hiding this comment.
.catch would be better.
Or
storage.put({
[connectionKeyString]: record,
[connectionRoomIndexString]: {},
}).catch(...)
There was a problem hiding this comment.
updated to multikey, great idea
| void storage.delete(connectionKeyString); | ||
| void storage.delete(connectionRoomIndexString); |
There was a problem hiding this comment.
delete also has a multi key version
There was a problem hiding this comment.
updated to multi key
| createConnectionKeyStringsGenerator(storage); | ||
| for await (const connectionKeyString of connectionKeyStringsGenerator) { | ||
| const connectionKey = must(connectionKeyFromString(connectionKeyString)); | ||
| void storage.put( |
There was a problem hiding this comment.
I was going for coallescing all of them to make it atomic. However, that can actually lead to OOM, as the coallesced writes can exceed memory limits. Added the await, which means the index building is not atomic, and can fail part of the way through... which is ok. We only update thee schema version meta to the new version if the migration completes. If it fails next time server starts we will retry migration.
Update AuthDO schema migration logic to deal with down-migrates (i.e. dealing with rolling back a deploy that does a migration). Also address arv's feedback from #491
Problem
Monday's revalidate connections cron is failing every time due to it exceeding memory limits. This is because the full list of connections is too large to fit in memory.
Solution
Refactor the two places where the AuthDO lists all connections (
_authRevalidateConnectionsand_authInvalidateAll), to instead incrementally load the connections.For
_authRevalidateConnectionswe need to group the connections by roomID, since we revalidate connections a room at a time by fetching the room's current connections. To do this incrementally without having to iterate the entire list of connections incrementally for each room necessitates creating a new index by room id over the connections. To handle the creation of this index for existings AuthDO's, an AuthDO Storage Schema Version is introduced and stored in the AuthDO's storage at keyauth_do_storage_schema_version. The AuthDO ensures the storage schema version is upgraded to the current version before processing any fetch.Fixes #487