Skip to content
This repository was archived by the owner on Apr 26, 2024. It is now read-only.

Make handle_new_client_event throws PartialStateConflictError #14665

Merged
MatMaul merged 14 commits into from
Dec 15, 2022
Merged

Make handle_new_client_event throws PartialStateConflictError #14665

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
2519142
Make `handle_new_client_event` throws `PartialStateConflictError`
Dec 9, 2022
abd9bd7
Remove retry from _send_events_for_new_room as it is a new local room
Dec 13, 2022
31ac5be
Deal with _upgrade_room
Dec 13, 2022
0d283e4
Merge remote-tracking branch 'origin/develop' into mv/new-client-even…
Dec 13, 2022
bbc57e7
Fix merge
Dec 13, 2022
28c930b
Switch from for loop to while
Dec 13, 2022
1c735a4
Revert "Switch from for loop to while"
Dec 14, 2022
bfa03d5
Tweak loops
Dec 14, 2022
7575b6d
Raise after last try
Dec 15, 2022
c07765c
Format
Dec 15, 2022
834fee5
Mistake
Dec 15, 2022
eabadd8
Add cache cleanup
Dec 15, 2022
95a3ae7
Update synapse/util/caches/response_cache.py
Dec 15, 2022
0cd4475
Raise instead of return
Dec 15, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/14665.misc
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Change `handle_new_client_event` signature so that a 429 does not reach clients on `PartialStateConflictError`, and internally retry when needed instead.
111 changes: 72 additions & 39 deletions synapse/handlers/federation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1343,32 +1343,50 @@ async def exchange_third_party_invite(
)

EventValidator().validate_builder(builder)
event, context = await self.event_creation_handler.create_new_client_event(
builder=builder
)

event, context = await self.add_display_name_to_third_party_invite(
room_version_obj, event_dict, event, context
)
# Try 2 times, the first one could fail with PartialStateConflictError
# in send_membership_event, cf comment in except block.
for _ in range(2):
try:
(
event,
context,
) = await self.event_creation_handler.create_new_client_event(
builder=builder
)

EventValidator().validate_new(event, self.config)
event, context = await self.add_display_name_to_third_party_invite(
room_version_obj, event_dict, event, context
)

# We need to tell the transaction queue to send this out, even
# though the sender isn't a local user.
event.internal_metadata.send_on_behalf_of = self.hs.hostname
EventValidator().validate_new(event, self.config)

try:
validate_event_for_room_version(event)
await self._event_auth_handler.check_auth_rules_from_context(event)
except AuthError as e:
logger.warning("Denying new third party invite %r because %s", event, e)
raise e
# We need to tell the transaction queue to send this out, even
# though the sender isn't a local user.
event.internal_metadata.send_on_behalf_of = self.hs.hostname

await self._check_signature(event, context)
try:
validate_event_for_room_version(event)
await self._event_auth_handler.check_auth_rules_from_context(
event
)
except AuthError as e:
logger.warning(
"Denying new third party invite %r because %s", event, e
)
raise e

# We retrieve the room member handler here as to not cause a cyclic dependency
member_handler = self.hs.get_room_member_handler()
await member_handler.send_membership_event(None, event, context)
await self._check_signature(event, context)

# We retrieve the room member handler here as to not cause a cyclic dependency
member_handler = self.hs.get_room_member_handler()
await member_handler.send_membership_event(None, event, context)

break
except PartialStateConflictError:
# Persisting couldn't happen because the room got un-partial stated
# in the meantime and context needs to be recomputed, so let's do so.
pass
else:
destinations = {x.split(":", 1)[-1] for x in (sender_user_id, room_id)}

Expand Down Expand Up @@ -1400,28 +1418,43 @@ async def on_exchange_third_party_invite_request(
room_version_obj, event_dict
)

event, context = await self.event_creation_handler.create_new_client_event(
builder=builder
)
event, context = await self.add_display_name_to_third_party_invite(
room_version_obj, event_dict, event, context
)
# Try 2 times, the first one could fail with PartialStateConflictError
# in send_membership_event, cf comment in except block.
for _ in range(2):
try:
(
event,
context,
) = await self.event_creation_handler.create_new_client_event(
builder=builder
)
event, context = await self.add_display_name_to_third_party_invite(
room_version_obj, event_dict, event, context
)

try:
validate_event_for_room_version(event)
await self._event_auth_handler.check_auth_rules_from_context(event)
except AuthError as e:
logger.warning("Denying third party invite %r because %s", event, e)
raise e
await self._check_signature(event, context)
try:
validate_event_for_room_version(event)
await self._event_auth_handler.check_auth_rules_from_context(event)
except AuthError as e:
logger.warning("Denying third party invite %r because %s", event, e)
raise e
await self._check_signature(event, context)

# We need to tell the transaction queue to send this out, even
# though the sender isn't a local user.
event.internal_metadata.send_on_behalf_of = get_domain_from_id(
event.sender
)

# We need to tell the transaction queue to send this out, even
# though the sender isn't a local user.
event.internal_metadata.send_on_behalf_of = get_domain_from_id(event.sender)
# We retrieve the room member handler here as to not cause a cyclic dependency
member_handler = self.hs.get_room_member_handler()
await member_handler.send_membership_event(None, event, context)

# We retrieve the room member handler here as to not cause a cyclic dependency
member_handler = self.hs.get_room_member_handler()
await member_handler.send_membership_event(None, event, context)
break
except PartialStateConflictError:
# Persisting couldn't happen because the room got un-partial stated
# in the meantime and context needs to be recomputed, so let's do so.
pass

async def add_display_name_to_third_party_invite(
self,
Expand Down
196 changes: 102 additions & 94 deletions synapse/handlers/message.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,6 @@
AuthError,
Codes,
ConsentNotGivenError,
LimitExceededError,
NotFoundError,
ShadowBanError,
SynapseError,
Expand Down Expand Up @@ -998,60 +997,70 @@ async def create_and_send_nonmember_event(
event.internal_metadata.stream_ordering,
)

event, context = await self.create_event(
requester,
event_dict,
txn_id=txn_id,
allow_no_prev_events=allow_no_prev_events,
prev_event_ids=prev_event_ids,
state_event_ids=state_event_ids,
outlier=outlier,
historical=historical,
depth=depth,
)
# Try 2 times, the first one could fail with PartialStateConflictError
# in handle_new_client_event, cf comment in except block.
for _ in range(2):
try:
event, context = await self.create_event(
requester,
event_dict,
txn_id=txn_id,
allow_no_prev_events=allow_no_prev_events,
prev_event_ids=prev_event_ids,
state_event_ids=state_event_ids,
outlier=outlier,
historical=historical,
depth=depth,
)

assert self.hs.is_mine_id(event.sender), "User must be our own: %s" % (
event.sender,
)
assert self.hs.is_mine_id(event.sender), "User must be our own: %s" % (
event.sender,
)

spam_check_result = await self.spam_checker.check_event_for_spam(event)
if spam_check_result != self.spam_checker.NOT_SPAM:
if isinstance(spam_check_result, tuple):
try:
[code, dict] = spam_check_result
raise SynapseError(
403,
"This message had been rejected as probable spam",
code,
dict,
)
except ValueError:
logger.error(
"Spam-check module returned invalid error value. Expecting [code, dict], got %s",
spam_check_result,
)
spam_check_result = await self.spam_checker.check_event_for_spam(event)
if spam_check_result != self.spam_checker.NOT_SPAM:
if isinstance(spam_check_result, tuple):
try:
[code, dict] = spam_check_result
raise SynapseError(
403,
"This message had been rejected as probable spam",
code,
dict,
)
except ValueError:
logger.error(
"Spam-check module returned invalid error value. Expecting [code, dict], got %s",
spam_check_result,
)

raise SynapseError(
403,
"This message has been rejected as probable spam",
Codes.FORBIDDEN,
)
raise SynapseError(
403,
"This message has been rejected as probable spam",
Codes.FORBIDDEN,
)

# Backwards compatibility: if the return value is not an error code, it
# means the module returned an error message to be included in the
# SynapseError (which is now deprecated).
raise SynapseError(
403,
spam_check_result,
Codes.FORBIDDEN,
# Backwards compatibility: if the return value is not an error code, it
# means the module returned an error message to be included in the
# SynapseError (which is now deprecated).
raise SynapseError(
403,
spam_check_result,
Codes.FORBIDDEN,
)

ev = await self.handle_new_client_event(
requester=requester,
events_and_context=[(event, context)],
ratelimit=ratelimit,
ignore_shadow_ban=ignore_shadow_ban,
)

ev = await self.handle_new_client_event(
requester=requester,
events_and_context=[(event, context)],
ratelimit=ratelimit,
ignore_shadow_ban=ignore_shadow_ban,
)
break
except PartialStateConflictError:
# Persisting couldn't happen because the room got un-partial stated
# in the meantime and context needs to be recomputed, so let's do so.
pass

# we know it was persisted, so must have a stream ordering
assert ev.internal_metadata.stream_ordering
Expand Down Expand Up @@ -1355,7 +1364,7 @@ async def handle_new_client_event(

Raises:
ShadowBanError if the requester has been shadow-banned.
SynapseError(503) if attempting to persist a partial state event in
PartialStateConflictError if attempting to persist a partial state event in
a room that has been un-partial stated.
"""
extra_users = extra_users or []
Expand Down Expand Up @@ -1417,34 +1426,23 @@ async def handle_new_client_event(
# We now persist the event (and update the cache in parallel, since we
# don't want to block on it).
event, context = events_and_context[0]
try:
result, _ = await make_deferred_yieldable(
gather_results(
(
run_in_background(
self._persist_events,
requester=requester,
events_and_context=events_and_context,
ratelimit=ratelimit,
extra_users=extra_users,
),
run_in_background(
self.cache_joined_hosts_for_events, events_and_context
).addErrback(
log_failure, "cache_joined_hosts_for_event failed"
),
result, _ = await make_deferred_yieldable(
gather_results(
(
run_in_background(
self._persist_events,
requester=requester,
events_and_context=events_and_context,
ratelimit=ratelimit,
extra_users=extra_users,
),
consumeErrors=True,
)
).addErrback(unwrapFirstError)
except PartialStateConflictError as e:
# The event context needs to be recomputed.
# Turn the error into a 429, as a hint to the client to try again.
logger.info(
"Room %s was un-partial stated while persisting client event.",
event.room_id,
run_in_background(
self.cache_joined_hosts_for_events, events_and_context
).addErrback(log_failure, "cache_joined_hosts_for_event failed"),
),
consumeErrors=True,
)
raise LimitExceededError(msg=e.msg, errcode=e.errcode, retry_after_ms=0)
).addErrback(unwrapFirstError)

return result

Expand Down Expand Up @@ -2005,26 +2003,36 @@ async def _send_dummy_event_for_room(self, room_id: str) -> bool:
for user_id in members:
requester = create_requester(user_id, authenticated_entity=self.server_name)
try:
event, context = await self.create_event(
requester,
{
"type": EventTypes.Dummy,
"content": {},
"room_id": room_id,
"sender": user_id,
},
)
# Try 2 times, the first one could fail with PartialStateConflictError
# in handle_new_client_event, cf comment in except block.
for _ in range(2):
try:
event, context = await self.create_event(
requester,
{
"type": EventTypes.Dummy,
"content": {},
"room_id": room_id,
"sender": user_id,
},
)

event.internal_metadata.proactively_send = False
event.internal_metadata.proactively_send = False

# Since this is a dummy-event it is OK if it is sent by a
# shadow-banned user.
await self.handle_new_client_event(
requester,
events_and_context=[(event, context)],
ratelimit=False,
ignore_shadow_ban=True,
)
# Since this is a dummy-event it is OK if it is sent by a
# shadow-banned user.
await self.handle_new_client_event(
requester,
events_and_context=[(event, context)],
ratelimit=False,
ignore_shadow_ban=True,
)

break
except PartialStateConflictError:
# Persisting couldn't happen because the room got un-partial stated
# in the meantime and context needs to be recomputed, so let's do so.
pass
return True
except AuthError:
logger.info(
Expand Down
Loading