nyeki/payphone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(bot): 🐛 Redis failing to connect in different event loops.

Browse Source
This commit is contained in:
Nyeki.py
2024-10-23 23:33:25 -03:00
parent 7acfb87910
commit c5143de3ba
11 changed files with 133 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+1
View File
@@ -5,6 +5,7 @@ COPY . .
RUN pip install uv
RUN uv sync --no-dev
RUN uv add uvloop
ENTRYPOINT [ "/bin/bash" ]
CMD [ "run.sh" ]
payphone/bot.py
+28 -13
View File
@@ -2,6 +2,8 @@ from typing import Literal
from datetime import datetime, timezone, timedelta
import sys
import discord
from discord import abc
from discord.ext import commands
@@ -9,12 +11,17 @@ from discord.ext.commands import errors
from tortoise import expressions, transactions, exceptions as tortoise_exc
import redis.asyncio as aredis
import asyncpg.exceptions
from payphone.db import setup, models
from payphone.core import constants, exceptions, formatting, help_command, config
from payphone.core.redis import redis
class Payphone(commands.AutoShardedBot):
redis: aredis.Redis
def __init__(self) -> None:
intents = discord.Intents.default()
intents.message_content = True
@@ -39,13 +46,11 @@ class Payphone(commands.AutoShardedBot):
console.
"""
self.redis = aredis.Redis(host="localhost" if config.IS_DEBUG else "redis")
await setup.init()
await redis.ping()
await redis.delete(
constants.CHANNELS_ONCALL_SET, constants.CHANNELS_SEARCH_QUEUE
)
await self.redis.ping()
print(f"Ready! Logged in as {self.user.display_name}.")
@@ -121,7 +126,7 @@ class Payphone(commands.AutoShardedBot):
user_id = user if isinstance(user, int) else user.id
key = constants.BAN_TEMPLATE % user_id
is_banned = await redis.get(key)
is_banned = await self.redis.get(key)
if is_banned is not None:
return bool(int(is_banned))
@@ -135,7 +140,7 @@ class Payphone(commands.AutoShardedBot):
),
).exists()
await redis.set(key, int(is_banned), ex=300)
await self.redis.set(key, int(is_banned), ex=300)
return is_banned
@@ -203,7 +208,7 @@ class Payphone(commands.AutoShardedBot):
report_id=report_id,
)
await redis.set(key, "1", ex=300)
await self.redis.set(key, "1", ex=300)
return row
@@ -230,7 +235,7 @@ class Payphone(commands.AutoShardedBot):
user_id=user_id,
).update(expires_at=datetime.now(timezone.utc))
await redis.set(key, "0", ex=300)
await self.redis.set(key, "0", ex=300)
async def get_prefix(
self, message: discord.Message | None = None, server_id: int | None = None
@@ -282,7 +287,7 @@ class Payphone(commands.AutoShardedBot):
key = constants.PREFIX_TEMPLATE % server_id
prefix: bytes | None = await redis.get(key)
prefix: bytes | None = await self.redis.get(key)
if prefix is not None:
prefix = prefix.decode()
@@ -297,7 +302,7 @@ class Payphone(commands.AutoShardedBot):
id=server_id,
)
await redis.set(key, server.prefix, ex=60 * 60)
await self.redis.set(key, server.prefix, ex=60 * 60)
prefix = server.prefix
@@ -329,7 +334,7 @@ class Payphone(commands.AutoShardedBot):
await models.Server.filter(id=server_id).update(prefix=prefix)
await redis.set(constants.PREFIX_TEMPLATE % server_id, prefix, ex=60 * 60)
await self.redis.set(constants.PREFIX_TEMPLATE % server_id, prefix, ex=60 * 60)
async def on_command_error(
self, context: commands.Context, exception: commands.CommandError
@@ -390,6 +395,16 @@ class Payphone(commands.AutoShardedBot):
else:
return await super().on_command_error(context, exception)
async def on_error(
self, event_method: str, *args: discord.Any, **kwargs: discord.Any
) -> None:
_, exception, _ = sys.exc_info()
if isinstance(exception, asyncpg.exceptions.InterfaceError):
return
return await super().on_error(event_method, *args, **kwargs)
@property
def embed_author(self) -> discord.EmbedAuthor:
"""The embed author for embeds sent by Payphone.
payphone/cogs/calls/cog.py
+45 -45
View File
@@ -20,7 +20,6 @@ from payphone.db import utils as dbutils, models
from payphone.cogs import base
from payphone.cogs.calls import embeds
from payphone.core import constants, checks, censorship, formatting, utils
from payphone.core.redis import redis
class Calls(base.Cog):
@@ -79,7 +78,7 @@ class Calls(base.Cog):
if len(hung_up_channel_ids) > 0:
await models.Channel.filter(id__in=hung_up_channel_ids).update(call_id=None)
await redis.srem(constants.CHANNELS_ONCALL_SET, *hung_up_channel_ids)
await self.bot.redis.srem(constants.CHANNELS_ONCALL_SET, *hung_up_channel_ids)
@commands.command(aliases=["c"])
@commands.guild_only()
@@ -97,7 +96,7 @@ class Calls(base.Cog):
user_id = ctx.author.id
channel_id = ctx.channel.id
is_on_call = await redis.sismember(
is_on_call = await self.bot.redis.sismember(
constants.CHANNELS_ONCALL_SET, str(channel_id)
)
@@ -107,7 +106,7 @@ class Calls(base.Cog):
)
return
is_searching = await redis.sismember(
is_searching = await self.bot.redis.sismember(
constants.CHANNELS_SEARCH_QUEUE, str(channel_id)
)
@@ -156,7 +155,7 @@ class Calls(base.Cog):
if last_calls_rss_channels_map[call_rs.call_id] != call_rs.channel_id:
last_calls_other_channels_ids.add(call_rs.channel_id)
other_channel_id: int = await redis.spop(constants.CHANNELS_SEARCH_QUEUE)
other_channel_id: int = await self.bot.redis.spop(constants.CHANNELS_SEARCH_QUEUE)
i = 0
while True:
@@ -165,7 +164,7 @@ class Calls(base.Cog):
other_channel_id = None
break
# Redis returns the ID as a bytes string.
# self.bot.redis returns the ID as a bytes string.
other_channel_id = int(other_channel_id)
if i == 5:
@@ -174,9 +173,9 @@ class Calls(base.Cog):
i += 1
if other_channel_id in last_calls_other_channels_ids:
await redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(other_channel_id))
await self.bot.redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(other_channel_id))
other_channel_id = await redis.spop(constants.CHANNELS_SEARCH_QUEUE)
other_channel_id = await self.bot.redis.spop(constants.CHANNELS_SEARCH_QUEUE)
if other_channel_id is None:
break
@@ -188,7 +187,7 @@ class Calls(base.Cog):
channel = await dbutils.get_channel(ctx.channel)
if other_channel_id is not None:
await redis.sadd(
await self.bot.redis.sadd(
constants.CHANNELS_ONCALL_SET, str(channel_id), str(other_channel_id)
)
@@ -217,12 +216,12 @@ class Calls(base.Cog):
).update(call=call)
except errors.DiscordException:
await redis.srem(
await self.bot.redis.srem(
constants.CHANNELS_ONCALL_SET,
str(channel_id),
str(other_channel_id),
)
await redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(channel_id))
await self.bot.redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(channel_id))
msg_phone_answered = ""
await ctx.send(
@@ -234,7 +233,7 @@ class Calls(base.Cog):
)
else:
await redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(channel_id))
await self.bot.redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(channel_id))
await ctx.send(
(
@@ -259,7 +258,7 @@ class Calls(base.Cog):
user_id = ctx.author.id
channel_id = ctx.channel.id
is_on_call = await redis.sismember(
is_on_call = await self.bot.redis.sismember(
constants.CHANNELS_ONCALL_SET, str(channel_id)
)
@@ -271,7 +270,7 @@ class Calls(base.Cog):
"channels"
)
await redis.sadd(constants.CALL_IDS_TO_FLUSH, call.id)
await self.bot.redis.sadd(constants.CALL_IDS_TO_FLUSH, call.id)
call.ended_at = datetime.now(timezone.utc)
await call.save()
@@ -290,7 +289,7 @@ class Calls(base.Cog):
except errors.DiscordException:
pass
await redis.srem(
await self.bot.redis.srem(
constants.CHANNELS_ONCALL_SET,
*(
[str(channel_id)]
@@ -337,7 +336,7 @@ class Calls(base.Cog):
if last_calls_rss_channels_map[call_rs.call_id] != call_rs.channel_id:
last_calls_other_channels_ids.add(call_rs.channel_id)
other_channel_id: int = await redis.spop(constants.CHANNELS_SEARCH_QUEUE)
other_channel_id: int = await self.bot.redis.spop(constants.CHANNELS_SEARCH_QUEUE)
i = 0
while True:
@@ -346,7 +345,7 @@ class Calls(base.Cog):
other_channel_id = None
break
# Redis returns the ID as a bytes string.
# self.bot.redis returns the ID as a bytes string.
other_channel_id = int(other_channel_id)
if i == 5:
@@ -355,9 +354,9 @@ class Calls(base.Cog):
i += 1
if other_channel_id in last_calls_other_channels_ids:
await redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(other_channel_id))
await self.bot.redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(other_channel_id))
other_channel_id = await redis.spop(constants.CHANNELS_SEARCH_QUEUE)
other_channel_id = await self.bot.redis.spop(constants.CHANNELS_SEARCH_QUEUE)
if other_channel_id is None:
break
@@ -367,7 +366,7 @@ class Calls(base.Cog):
break
if other_channel_id is not None:
await redis.sadd(
await self.bot.redis.sadd(
constants.CHANNELS_ONCALL_SET, str(channel_id), str(other_channel_id)
)
@@ -396,12 +395,12 @@ class Calls(base.Cog):
).update(call=call)
except errors.DiscordException:
await redis.srem(
await self.bot.redis.srem(
constants.CHANNELS_ONCALL_SET,
str(channel_id),
str(other_channel_id),
)
await redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(channel_id))
await self.bot.redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(channel_id))
msg_phone_answered = ""
await ctx.send(
@@ -417,7 +416,7 @@ class Calls(base.Cog):
)
else:
await redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(channel_id))
await self.bot.redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(channel_id))
channel.call = None
await channel.save()
@@ -441,7 +440,7 @@ class Calls(base.Cog):
channel_id = ctx.channel.id
is_on_call = await redis.sismember(
is_on_call = await self.bot.redis.sismember(
constants.CHANNELS_ONCALL_SET, str(channel_id)
)
@@ -452,13 +451,13 @@ class Calls(base.Cog):
"channels"
)
await redis.sadd(constants.CALL_IDS_TO_FLUSH, call.id)
await self.bot.redis.sadd(constants.CALL_IDS_TO_FLUSH, call.id)
other_channel: models.Channel = find(
lambda c: c.id != channel_id, call.channels
)
await redis.srem(
await self.bot.redis.srem(
constants.CHANNELS_ONCALL_SET,
*(
[str(channel_id)]
@@ -489,12 +488,12 @@ class Calls(base.Cog):
call.ended_at = datetime.now(timezone.utc)
await call.save()
is_searching = await redis.sismember(
is_searching = await self.bot.redis.sismember(
constants.CHANNELS_SEARCH_QUEUE, str(channel_id)
)
if is_searching:
await redis.srem(constants.CHANNELS_SEARCH_QUEUE, str(channel_id))
await self.bot.redis.srem(constants.CHANNELS_SEARCH_QUEUE, str(channel_id))
await ctx.send(f"☎️ {secrets.choice(constants.MSGS_PHONE_NO_ANSWER)}")
@@ -577,7 +576,7 @@ class Calls(base.Cog):
call.ended_at = datetime.now(timezone.utc)
await call.save()
await redis.sadd(constants.CALL_IDS_TO_FLUSH, call.id)
await self.bot.redis.sadd(constants.CALL_IDS_TO_FLUSH, call.id)
updated_channels = []
@@ -587,7 +586,7 @@ class Calls(base.Cog):
await models.Channel.bulk_update(updated_channels, ["call_id"])
await redis.srem(
await self.bot.redis.srem(
constants.CHANNELS_ONCALL_SET,
str(updated_channels[0]),
str(updated_channels[1]),
@@ -597,7 +596,7 @@ class Calls(base.Cog):
other_channel = find(lambda c: c.id != channel_id, updated_channels)
await redis.srem(
await self.bot.redis.srem(
constants.CHANNELS_ONCALL_SET, str(other_channel.id), str(channel_id)
)
@@ -666,7 +665,7 @@ class Calls(base.Cog):
user_id = message.author.id
channel_id = message.channel.id
is_on_call = await redis.sismember(
is_on_call = await self.bot.redis.sismember(
constants.CHANNELS_ONCALL_SET, str(channel_id)
)
@@ -686,7 +685,7 @@ class Calls(base.Cog):
)
except exceptions.DoesNotExist:
await redis.srem(constants.CHANNELS_ONCALL_SET, str(channel_id))
await self.bot.redis.srem(constants.CHANNELS_ONCALL_SET, str(channel_id))
return
other_channel: models.Channel = find(
@@ -755,6 +754,7 @@ class Calls(base.Cog):
if message.reference:
embed, replying_message = await embeds.ReplyEmbed(
self.bot.redis,
message.reference, call_id=call.id
)
@@ -769,7 +769,7 @@ class Calls(base.Cog):
wait=True,
)
pipe = redis.pipeline()
pipe = self.bot.redis.pipeline()
pipe.hset(
constants.CALL_MESSAGES_TEMPLATE % call.id,
@@ -815,7 +815,7 @@ class Calls(base.Cog):
)
if replying_to_id is not None:
other_replies = await redis.hget(
other_replies = await self.bot.redis.hget(
constants.CALL_MESSAGE_REPLIES_TEMPLATE % call.id, replying_to_id
)
@@ -858,7 +858,7 @@ class Calls(base.Cog):
if channel.call_id is None:
return
message_json = await redis.hget(
message_json = await self.bot.redis.hget(
constants.CALL_MESSAGES_TEMPLATE % channel.call_id, message_id
)
@@ -871,7 +871,7 @@ class Calls(base.Cog):
message_dict["was_deleted"] = True
await redis.hset(
await self.bot.redis.hset(
constants.CALL_MESSAGES_TEMPLATE % channel.call_id,
message.id,
orjson.dumps(message_dict),
@@ -892,7 +892,7 @@ class Calls(base.Cog):
# This message was already deleted.
pass
message_replies = await redis.hget(
message_replies = await self.bot.redis.hget(
constants.CALL_MESSAGE_REPLIES_TEMPLATE % channel.call_id, message.id
)
@@ -937,7 +937,7 @@ class Calls(base.Cog):
if channel.call_id is None:
return
message_json = await redis.hget(
message_json = await self.bot.redis.hget(
constants.CALL_MESSAGES_TEMPLATE % channel.call_id, message_id
)
@@ -967,13 +967,13 @@ class Calls(base.Cog):
blocked_word = censorship.moderate(content)
iteration = int(
await redis.hget(
await self.bot.redis.hget(
constants.CALL_MESSAGE_CONTENT_ITERATIONS_TEMPLATE % channel.call_id,
message.id,
)
)
pipe = redis.pipeline()
pipe = self.bot.redis.pipeline()
pipe.hincrby(
constants.CALL_MESSAGE_CONTENT_ITERATIONS_TEMPLATE % channel.call_id,
@@ -1031,13 +1031,13 @@ class Calls(base.Cog):
call_id = channel.call_id
other_channel_id = None
other_channel_message_id = await redis.hget(
other_channel_message_id = await self.bot.redis.hget(
constants.CALL_MESSAGE_FROM_PROPAGATION_TEMPLATE % call_id,
payload.message_id,
)
if other_channel_message_id is None:
other_channel_message_json = await redis.hget(
other_channel_message_json = await self.bot.redis.hget(
constants.CALL_MESSAGES_TEMPLATE % call_id, payload.message_id
)
@@ -1093,13 +1093,13 @@ class Calls(base.Cog):
call_id = channel.call_id
other_channel_id = None
other_channel_message_id = await redis.hget(
other_channel_message_id = await self.bot.redis.hget(
constants.CALL_MESSAGE_FROM_PROPAGATION_TEMPLATE % call_id,
payload.message_id,
)
if other_channel_message_id is None:
other_channel_message_json = await redis.hget(
other_channel_message_json = await self.bot.redis.hget(
constants.CALL_MESSAGES_TEMPLATE % call_id, payload.message_id
)
payphone/cogs/calls/embeds.py
+8 -4
View File
@@ -2,17 +2,19 @@ import discord
import orjson
import redis.asyncio as aredis
from payphone.db import models
from payphone.core import constants, formatting
from payphone.core.redis import redis
async def ReplyEmbed(
reference: discord.MessageReference, call_id: int
redis: aredis.Redis, reference: discord.MessageReference, call_id: int
) -> tuple[discord.Embed | None, models.Message | None]:
"""Generates the embed sent on calls to represent replies.
Args:
redis (redis.asyncio.Redis): An instance of a Redis connection handler.
reference (discord.MessageReference): The message reference.
call_id (int): The ID of the call the message was sent in.
@@ -43,9 +45,11 @@ async def ReplyEmbed(
message = models.Message(**orjson.loads(unflushed_message))
unflushed_message_content_iteration = int(await redis.hget(
unflushed_message_content_iteration = int(
await redis.hget(
constants.CALL_MESSAGE_CONTENT_ITERATIONS_TEMPLATE % call_id, message.id
))
)
)
unflushed_message_content = await redis.hget(
constants.CALL_MESSAGE_CONTENTS_TEMPLATE % call_id,
constants.CALL_MESSAGE_CONTENTS_HASH_TEMPLATE
payphone/cogs/mods/cog.py
+8 -10
View File
@@ -9,7 +9,6 @@ import discord
from discord import errors
from discord.ext import commands
import tortoise
from tortoise import transactions, expressions
from procs import flush_messages
@@ -19,7 +18,6 @@ from payphone.db import utils, models
from payphone.cogs import base
from payphone.cogs.mods import embeds, views
from payphone.core import checks, converters, constants, formatting as fmt
from payphone.core.redis import redis
class Mods(base.Cog):
@@ -245,9 +243,9 @@ class Mods(base.Cog):
- `[<reason>]`: The reason of the shutdown.
"""
oncall_channel_ids = list(await redis.smembers(constants.CHANNELS_ONCALL_SET))
oncall_channel_ids = list(await self.bot.redis.smembers(constants.CHANNELS_ONCALL_SET))
searching_channel_ids = list(
await redis.smembers(constants.CHANNELS_SEARCH_QUEUE)
await self.bot.redis.smembers(constants.CHANNELS_SEARCH_QUEUE)
)
channel_ids = [int(c) for c in oncall_channel_ids] + [
@@ -272,9 +270,9 @@ class Mods(base.Cog):
await asyncio.sleep(seconds)
oncall_channel_ids = list(await redis.smembers(constants.CHANNELS_ONCALL_SET))
oncall_channel_ids = list(await self.bot.redis.smembers(constants.CHANNELS_ONCALL_SET))
searching_channel_ids = list(
await redis.smembers(constants.CHANNELS_SEARCH_QUEUE)
await self.bot.redis.smembers(constants.CHANNELS_SEARCH_QUEUE)
)
channel_ids = [int(c) for c in oncall_channel_ids] + [
@@ -296,13 +294,13 @@ class Mods(base.Cog):
"⚡ Notified all channels of calls hang up. Flushing messages to the DB..."
)
await redis.delete(constants.CHANNELS_ONCALL_SET)
await self.bot.redis.delete(constants.CHANNELS_ONCALL_SET)
calls = await models.Call.filter(ended_at=None)
call_ids = [call.id for call in calls]
if call_ids:
await redis.sadd(constants.CALL_IDS_TO_FLUSH, *call_ids)
await self.bot.redis.sadd(constants.CALL_IDS_TO_FLUSH, *call_ids)
await models.Call.filter(ended_at=None).update(
ended_at=datetime.now(timezone.utc)
@@ -330,11 +328,11 @@ class Mods(base.Cog):
stats = [
(
"On-call Channels",
lambda: redis.scard(constants.CHANNELS_ONCALL_SET),
lambda: self.bot.redis.scard(constants.CHANNELS_ONCALL_SET),
),
(
"Searching Channels",
lambda: redis.scard(constants.CHANNELS_SEARCH_QUEUE),
lambda: self.bot.redis.scard(constants.CHANNELS_SEARCH_QUEUE),
),
(
"Messages",
payphone/core/checks.py
+1 -2
View File
@@ -2,7 +2,6 @@ from discord.ext import commands
from payphone.db import utils
from payphone.core import exceptions, constants
from payphone.core.redis import redis
def is_moderator():
@@ -40,7 +39,7 @@ def is_on_call():
async def predicate(ctx: commands.Context) -> bool:
channel_id = ctx.channel.id
is_on_call = await redis.sismember(
is_on_call = await ctx.bot.redis.sismember(
constants.CHANNELS_ONCALL_SET, str(channel_id)
)
payphone/core/converters.py
+2 -3
View File
@@ -9,7 +9,6 @@ import orjson
from payphone.db import models
from payphone.core import exceptions, constants
from payphone.core.redis import redis
class Timedelta(converter.Converter):
@@ -82,7 +81,7 @@ class DiscordUserID(converter.Converter):
if channel is None:
raise converter.BadArgument()
original_message_id = await redis.hget(
original_message_id = await ctx.bot.redis.hget(
constants.CALL_MESSAGE_FROM_PROPAGATION_TEMPLATE
% channel.call_id,
ref.message_id,
@@ -91,7 +90,7 @@ class DiscordUserID(converter.Converter):
if original_message_id is None:
raise converter.BadArgument()
original_message_json = await redis.hget(
original_message_json = await ctx.bot.redis.hget(
constants.CALL_MESSAGES_TEMPLATE % channel.call_id,
int(original_message_id),
)
payphone/core/redis.py
-6
View File
@@ -1,6 +0,0 @@
from redis import asyncio as aredis
from payphone.core import config
redis = aredis.Redis(host="localhost" if config.IS_DEBUG else "redis")
procs/flush_messages.py
+5 -2
View File
@@ -3,10 +3,13 @@ async def flush():
import asyncpg.exceptions
import redis.asyncio as aredis
from payphone.db import models
from payphone.core import constants, utils
from payphone.core.redis import redis
from payphone.core import constants, utils, config
redis = aredis.Redis(host="localhost" if config.IS_DEBUG else "redis")
pipe = redis.pipeline()
run.py
+31 -1
View File
@@ -15,6 +15,16 @@ def shard(shard_id: int, shard_count: int) -> None:
bot.shard_ids = [shard_id]
bot.shard_count = shard_count
try:
import uvloop # pylint: disable=import-error
bot.loop = uvloop.new_event_loop()
print(f"Running shard {shard_id} with uvloop.")
except ImportError:
print("Running with default loop.")
finally:
bot.run(config.TOKEN)
@@ -23,6 +33,23 @@ def flush_process():
from procs.flush_messages import loop
try:
import uvloop # pylint: disable=import-error
uvloop.install()
print("Flushing messages with uvloop.")
except ImportError:
try:
import winloop
winloop.install()
print("Flushing messages with winloop.")
except ImportError:
print("Could not find winloop nor uvloop for messages flushing.")
finally:
asyncio.run(loop())
@@ -35,9 +62,12 @@ if __name__ == "__main__":
import httpx
import redis.asyncio as aredis
from payphone.db import setup, models
from payphone.core import config
from payphone.core.redis import redis
redis = aredis.Redis(host="localhost" if config.IS_DEBUG else "redis")
async def init():
await setup.init()
run.sh
+1 -1
View File
@@ -6,4 +6,4 @@ do
echo "Retrying migrations..."
done
uv run --no-dev run.py
uv run --no-dev python3 -u run.py