Race condition with pre-existing rows.

Author: karel26

core/pubsub.py

@@ -149,6 +149,19 @@ async def subscribe(self):
                 async with await AsyncConnection.connect(self.url, autocommit=True) as conn:
                     async with conn.cursor() as cursor:
                         # preprocess all rows that might be there
+                        async for row in await cursor.execute(sql.SQL("""
+                            WITH to_delete AS (
+                                SELECT id
+                                FROM   {table}
+                                WHERE  guild_id = %s
+                                  AND  node = %s
+                                ORDER BY id
+                            )
+                            DELETE FROM {table}
+                            WHERE id IN (SELECT id FROM to_delete)
+                            RETURNING id
+                        """).format(table=sql.Identifier(self.name)), (self.node.guild_id, self.node.name)):
+                            self.read_queue.put_nowait(row['id'])
                         await cursor.execute(sql.SQL("LISTEN {table}").format(table=sql.Identifier(self.name)))
                         gen = conn.notifies()
                         async for n in gen: