Skip to content

Commit

Permalink
Reduce volume of 'Waiting for current token' logs, which were introdu…
Browse files Browse the repository at this point in the history
…ced in v1.109.0. (#17428)

Introduced in: #17215

This caused us a minor bit of grief as the volume of logs produced was
much higher than normal

---------

Signed-off-by: Olivier 'reivilibre <[email protected]>
  • Loading branch information
reivilibre authored Jul 23, 2024
1 parent a9ee832 commit 1daae43
Show file tree
Hide file tree
Showing 3 changed files with 32 additions and 5 deletions.
1 change: 1 addition & 0 deletions changelog.d/17428.bugfix
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Reduce volume of 'Waiting for current token' logs, which were introduced in v1.109.0.
13 changes: 8 additions & 5 deletions synapse/notifier.py
Original file line number Diff line number Diff line change
Expand Up @@ -773,6 +773,7 @@ async def wait_for_stream_token(self, stream_token: StreamToken) -> bool:
stream_token = await self.event_sources.bound_future_token(stream_token)

start = self.clock.time_msec()
logged = False
while True:
current_token = self.event_sources.get_current_token()
if stream_token.is_before_or_eq(current_token):
Expand All @@ -783,11 +784,13 @@ async def wait_for_stream_token(self, stream_token: StreamToken) -> bool:
if now - start > 10_000:
return False

logger.info(
"Waiting for current token to reach %s; currently at %s",
stream_token,
current_token,
)
if not logged:
logger.info(
"Waiting for current token to reach %s; currently at %s",
stream_token,
current_token,
)
logged = True

# TODO: be better
await self.clock.sleep(0.5)
Expand Down
23 changes: 23 additions & 0 deletions synapse/types/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -777,6 +777,13 @@ def bound_stream_token(self, max_stream: int) -> "RoomStreamToken":

return super().bound_stream_token(max_stream)

def __str__(self) -> str:
instances = ", ".join(f"{k}: {v}" for k, v in sorted(self.instance_map.items()))
return (
f"RoomStreamToken(stream: {self.stream}, topological: {self.topological}, "
f"instances: {{{instances}}})"
)


@attr.s(frozen=True, slots=True, order=False)
class MultiWriterStreamToken(AbstractMultiWriterStreamToken):
Expand Down Expand Up @@ -873,6 +880,13 @@ def is_stream_position_in_range(

return True

def __str__(self) -> str:
instances = ", ".join(f"{k}: {v}" for k, v in sorted(self.instance_map.items()))
return (
f"MultiWriterStreamToken(stream: {self.stream}, "
f"instances: {{{instances}}})"
)


class StreamKeyType(Enum):
"""Known stream types.
Expand Down Expand Up @@ -1131,6 +1145,15 @@ def is_before_or_eq(self, other_token: "StreamToken") -> bool:

return True

def __str__(self) -> str:
return (
f"StreamToken(room: {self.room_key}, presence: {self.presence_key}, "
f"typing: {self.typing_key}, receipt: {self.receipt_key}, "
f"account_data: {self.account_data_key}, push_rules: {self.push_rules_key}, "
f"to_device: {self.to_device_key}, device_list: {self.device_list_key}, "
f"groups: {self.groups_key}, un_partial_stated_rooms: {self.un_partial_stated_rooms_key})"
)


StreamToken.START = StreamToken(
RoomStreamToken(stream=0), 0, 0, MultiWriterStreamToken(stream=0), 0, 0, 0, 0, 0, 0
Expand Down

0 comments on commit 1daae43

Please sign in to comment.