Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
0ce5692
Update delayed events to support no tokens
Half-Shot Oct 27, 2025
eef7045
More support for unauthed delayed event management
AndrewFerr Oct 28, 2025
7a34e62
Test delayed event management without auth
AndrewFerr Oct 29, 2025
711a6c2
Split delayed event management endpoints
AndrewFerr Oct 29, 2025
0cf0036
Commonize test function
AndrewFerr Oct 29, 2025
5418998
Merge with 'develop'
AndrewFerr Oct 29, 2025
cf2972a
Restore tests on common delay evt mgmt endpoint
AndrewFerr Oct 29, 2025
62b1c56
Use assert keyword to assist mypy
AndrewFerr Oct 29, 2025
1f84c5f
Merge with 'develop'
AndrewFerr Nov 7, 2025
a24da1f
Update schema change comment
AndrewFerr Nov 7, 2025
754083f
Add changelog
AndrewFerr Nov 7, 2025
422e843
Rename copied table after index creation
AndrewFerr Nov 7, 2025
f15b04d
Merge with 'develop'
AndrewFerr Nov 11, 2025
87752c9
Add UNIQUE INDEX instead of replacing pkey
AndrewFerr Nov 11, 2025
fa20174
Add unique constraint in background
AndrewFerr Nov 11, 2025
76872b1
Fix parameterized.expand for trial-olddeps
AndrewFerr Nov 11, 2025
43c5151
Pass generator into parameterized.expand, not list
AndrewFerr Nov 11, 2025
a8986e6
Include DelayedEventsStore in migrated DB stores
AndrewFerr Nov 3, 2025
f41a897
Add ordering info to background update
AndrewFerr Nov 11, 2025
a61db99
Move schema update comment out of sql file
AndrewFerr Nov 11, 2025
0e47cc5
Merge branch 'develop' into delayed-event-management-changes
AndrewFerr Nov 12, 2025
0d84673
Merge branch 'develop' into delayed-event-management-changes
AndrewFerr Nov 13, 2025
4cae1a0
Merge branch 'develop' into delayed-event-management-changes
AndrewFerr Nov 13, 2025
f35ff0d
Revert "Fix parameterized.expand for trial-olddeps"
AndrewFerr Nov 13, 2025
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/19152.feature
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Remove authentication from `POST /_matrix/client/v1/delayed_events`, and allow calling this endpoint with the update action to take (`send`/`cancel`/`restart`) in the request path instead of the body.
2 changes: 2 additions & 0 deletions synapse/_scripts/synapse_port_db.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@
from synapse.storage.databases.main import FilteringWorkerStore
from synapse.storage.databases.main.account_data import AccountDataWorkerStore
from synapse.storage.databases.main.client_ips import ClientIpBackgroundUpdateStore
from synapse.storage.databases.main.delayed_events import DelayedEventsStore
from synapse.storage.databases.main.deviceinbox import DeviceInboxBackgroundUpdateStore
from synapse.storage.databases.main.devices import DeviceBackgroundUpdateStore
from synapse.storage.databases.main.e2e_room_keys import EndToEndRoomKeyBackgroundStore
Expand Down Expand Up @@ -273,6 +274,7 @@ class Store(
RelationsWorkerStore,
EventFederationWorkerStore,
SlidingSyncStore,
DelayedEventsStore,
):
def execute(self, f: Callable[..., R], *args: Any, **kwargs: Any) -> Awaitable[R]:
return self.db_pool.runInteraction(f.__name__, f, *args, **kwargs)
Expand Down
64 changes: 14 additions & 50 deletions synapse/handlers/delayed_events.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
from synapse.api.errors import ShadowBanError, SynapseError
from synapse.api.ratelimiting import Ratelimiter
from synapse.config.workers import MAIN_PROCESS_INSTANCE_NAME
from synapse.http.site import SynapseRequest
from synapse.logging.context import make_deferred_yieldable
from synapse.logging.opentracing import set_tag
from synapse.metrics import SERVER_NAME_LABEL, event_processing_positions
Expand All @@ -29,11 +30,9 @@
)
from synapse.storage.databases.main.delayed_events import (
DelayedEventDetails,
DelayID,
EventType,
StateKey,
Timestamp,
UserLocalpart,
)
from synapse.storage.databases.main.state_deltas import StateDelta
from synapse.types import (
Expand Down Expand Up @@ -399,96 +398,63 @@ def on_added(self, next_send_ts: int) -> None:
if self._next_send_ts_changed(next_send_ts):
self._schedule_next_at(next_send_ts)

async def cancel(self, requester: Requester, delay_id: str) -> None:
async def cancel(self, request: SynapseRequest, delay_id: str) -> None:
"""
Cancels the scheduled delivery of the matching delayed event.

Args:
requester: The owner of the delayed event to act on.
delay_id: The ID of the delayed event to act on.

Raises:
NotFoundError: if no matching delayed event could be found.
"""
assert self._is_master
await self._delayed_event_mgmt_ratelimiter.ratelimit(
requester,
(requester.user.to_string(), requester.device_id),
None, request.getClientAddress().host
)
await make_deferred_yieldable(self._initialized_from_db)

next_send_ts = await self._store.cancel_delayed_event(
delay_id=delay_id,
user_localpart=requester.user.localpart,
)
next_send_ts = await self._store.cancel_delayed_event(delay_id)

if self._next_send_ts_changed(next_send_ts):
self._schedule_next_at_or_none(next_send_ts)

async def restart(self, requester: Requester, delay_id: str) -> None:
async def restart(self, request: SynapseRequest, delay_id: str) -> None:
"""
Restarts the scheduled delivery of the matching delayed event.

Args:
requester: The owner of the delayed event to act on.
delay_id: The ID of the delayed event to act on.

Raises:
NotFoundError: if no matching delayed event could be found.
"""
assert self._is_master
await self._delayed_event_mgmt_ratelimiter.ratelimit(
requester,
(requester.user.to_string(), requester.device_id),
None, request.getClientAddress().host
)
await make_deferred_yieldable(self._initialized_from_db)

next_send_ts = await self._store.restart_delayed_event(
delay_id=delay_id,
user_localpart=requester.user.localpart,
current_ts=self._get_current_ts(),
delay_id, self._get_current_ts()
)

if self._next_send_ts_changed(next_send_ts):
self._schedule_next_at(next_send_ts)

async def send(self, requester: Requester, delay_id: str) -> None:
async def send(self, request: SynapseRequest, delay_id: str) -> None:
"""
Immediately sends the matching delayed event, instead of waiting for its scheduled delivery.

Args:
requester: The owner of the delayed event to act on.
delay_id: The ID of the delayed event to act on.

Raises:
NotFoundError: if no matching delayed event could be found.
"""
assert self._is_master
# Use standard request limiter for sending delayed events on-demand,
# as an on-demand send is similar to sending a regular event.
await self._request_ratelimiter.ratelimit(requester)
await self._delayed_event_mgmt_ratelimiter.ratelimit(
None, request.getClientAddress().host
)
await make_deferred_yieldable(self._initialized_from_db)

event, next_send_ts = await self._store.process_target_delayed_event(
delay_id=delay_id,
user_localpart=requester.user.localpart,
)
event, next_send_ts = await self._store.process_target_delayed_event(delay_id)

if self._next_send_ts_changed(next_send_ts):
self._schedule_next_at_or_none(next_send_ts)

await self._send_event(
DelayedEventDetails(
delay_id=DelayID(delay_id),
user_localpart=UserLocalpart(requester.user.localpart),
room_id=event.room_id,
type=event.type,
state_key=event.state_key,
origin_server_ts=event.origin_server_ts,
content=event.content,
device_id=event.device_id,
)
)
await self._send_event(event)

async def _send_on_timeout(self) -> None:
self._next_delayed_event_call = None
Expand Down Expand Up @@ -611,9 +577,7 @@ async def _send_event(
finally:
# TODO: If this is a temporary error, retry. Otherwise, consider notifying clients of the failure
try:
await self._store.delete_processed_delayed_event(
event.delay_id, event.user_localpart
)
await self._store.delete_processed_delayed_event(event.delay_id)
except Exception:
logger.exception("Failed to delete processed delayed event")

Expand Down
66 changes: 60 additions & 6 deletions synapse/rest/client/delayed_events.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,14 +47,11 @@ class UpdateDelayedEventServlet(RestServlet):

def __init__(self, hs: "HomeServer"):
super().__init__()
self.auth = hs.get_auth()
self.delayed_events_handler = hs.get_delayed_events_handler()

async def on_POST(
self, request: SynapseRequest, delay_id: str
) -> tuple[int, JsonDict]:
requester = await self.auth.get_user_by_req(request)

body = parse_json_object_from_request(request)
try:
action = str(body["action"])
Expand All @@ -75,11 +72,65 @@ async def on_POST(
)

if enum_action == _UpdateDelayedEventAction.CANCEL:
await self.delayed_events_handler.cancel(requester, delay_id)
await self.delayed_events_handler.cancel(request, delay_id)
elif enum_action == _UpdateDelayedEventAction.RESTART:
await self.delayed_events_handler.restart(requester, delay_id)
await self.delayed_events_handler.restart(request, delay_id)
elif enum_action == _UpdateDelayedEventAction.SEND:
await self.delayed_events_handler.send(requester, delay_id)
await self.delayed_events_handler.send(request, delay_id)
return 200, {}


class CancelDelayedEventServlet(RestServlet):
PATTERNS = client_patterns(
r"/org\.matrix\.msc4140/delayed_events/(?P<delay_id>[^/]+)/cancel$",
releases=(),
)
CATEGORY = "Delayed event management requests"

def __init__(self, hs: "HomeServer"):
super().__init__()
self.delayed_events_handler = hs.get_delayed_events_handler()

async def on_POST(
self, request: SynapseRequest, delay_id: str
) -> tuple[int, JsonDict]:
await self.delayed_events_handler.cancel(request, delay_id)
return 200, {}


class RestartDelayedEventServlet(RestServlet):
PATTERNS = client_patterns(
r"/org\.matrix\.msc4140/delayed_events/(?P<delay_id>[^/]+)/restart$",
releases=(),
)
CATEGORY = "Delayed event management requests"

def __init__(self, hs: "HomeServer"):
super().__init__()
self.delayed_events_handler = hs.get_delayed_events_handler()

async def on_POST(
self, request: SynapseRequest, delay_id: str
) -> tuple[int, JsonDict]:
await self.delayed_events_handler.restart(request, delay_id)
return 200, {}


class SendDelayedEventServlet(RestServlet):
PATTERNS = client_patterns(
r"/org\.matrix\.msc4140/delayed_events/(?P<delay_id>[^/]+)/send$",
releases=(),
)
CATEGORY = "Delayed event management requests"

def __init__(self, hs: "HomeServer"):
super().__init__()
self.delayed_events_handler = hs.get_delayed_events_handler()

async def on_POST(
self, request: SynapseRequest, delay_id: str
) -> tuple[int, JsonDict]:
await self.delayed_events_handler.send(request, delay_id)
return 200, {}


Expand Down Expand Up @@ -108,4 +159,7 @@ def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
# The following can't currently be instantiated on workers.
if hs.config.worker.worker_app is None:
UpdateDelayedEventServlet(hs).register(http_server)
CancelDelayedEventServlet(hs).register(http_server)
RestartDelayedEventServlet(hs).register(http_server)
SendDelayedEventServlet(hs).register(http_server)
DelayedEventsServlet(hs).register(http_server)
Loading
Loading