Send USER_IP commands on a different Redis channel, in order to reduce traffic to workers that do not process these commands.

changelog.d/12672.feature

@@ -0,0 +1 @@
+Send `USER_IP` commands on a different Redis channel, in order to reduce traffic to workers that do not process these commands.

changelog.d/12809.feature

@@ -0,0 +1 @@
+Send `USER_IP` commands on a different Redis channel, in order to reduce traffic to workers that do not process these commands.

synapse/replication/tcp/commands.py

@@ -58,6 +58,15 @@ def get_logcontext_id(self) -> str:
  # by default, we just use the command name.
  return self.NAME
 
+ def redis_channel_name(self, prefix: str) -> str:
+ """
+ Returns the Redis channel name upon which to publish this command.
+
+ Args:
+ prefix: The prefix for the channel.
+ """
+ return prefix
+
 
 SC = TypeVar("SC", bound="_SimpleCommand")
 
@@ -395,6 +404,9 @@ def __repr__(self) -> str:
  f"{self.user_agent!r}, {self.device_id!r}, {self.last_seen})"
  )
 
+ def redis_channel_name(self, prefix: str) -> str:
+ return f"{prefix}/USER_IP"
+
 
 class RemoteServerUpCommand(_SimpleCommand):
  """Sent when a worker has detected that a remote server is no longer

synapse/replication/tcp/redis.py

@@ -221,10 +221,10 @@ async def _async_send_command(self, cmd: Command) -> None:
  # remote instances.
  tcp_outbound_commands_counter.labels(cmd.NAME, "redis").inc()
 
+ channel_name = cmd.redis_channel_name(self.synapse_stream_prefix)
+
  await make_deferred_yieldable(
- self.synapse_outbound_redis_connection.publish(
- self.synapse_stream_prefix, encoded_string
- )
+ self.synapse_outbound_redis_connection.publish(channel_name, encoded_string)
  )