dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-08-28 19:08:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

Allow clients to send sticky events

Browse Source 
Including delayed events
This commit is contained in:
Olivier 'reivilibre
2026-01-16 09:00:00 +00:00
parent 1662a5c138
commit a5c56b2340
3 changed files with 40 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/handlers/delayed_events.py
+10 -3
View File
@@ -17,7 +17,7 @@ from typing import TYPE_CHECKING, Optional
from twisted.internet.interfaces import IDelayedCall
from synapse.api.constants import EventTypes
from synapse.api.constants import EventTypes, StickyEvent
from synapse.api.errors import ShadowBanError, SynapseError
from synapse.api.ratelimiting import Ratelimiter
from synapse.config.workers import MAIN_PROCESS_INSTANCE_NAME
@@ -333,6 +333,7 @@ class DelayedEventsHandler:
origin_server_ts: int | None,
content: JsonDict,
delay: int,
sticky_duration_ms: int | None,
) -> str:
"""
Creates a new delayed event and schedules its delivery.
@@ -346,7 +347,9 @@ class DelayedEventsHandler:
If None, the timestamp will be the actual time when the event is sent.
content: The content of the event to be sent.
delay: How long (in milliseconds) to wait before automatically sending the event.
sticky_duration_ms: If an MSC4354 sticky event: the sticky duration (in milliseconds).
The event will be attempted to be reliably delivered to clients and remote servers
during its sticky period.
Returns: The ID of the added delayed event.
Raises:
@@ -382,6 +385,7 @@ class DelayedEventsHandler:
origin_server_ts=origin_server_ts,
content=content,
delay=delay,
sticky_duration_ms=sticky_duration_ms,
)
if self._repl_client is not None:
@@ -570,7 +574,10 @@ class DelayedEventsHandler:
if event.state_key is not None:
event_dict["state_key"] = event.state_key
if event.sticky_duration_ms is not None:
event_dict[StickyEvent.FIELD_NAME] = {
"duration_ms": event.sticky_duration_ms,
}
(
sent_event,
_,
synapse/rest/client/room.py
+22 -1
View File
@@ -34,7 +34,7 @@ from prometheus_client.core import Histogram
from twisted.web.server import Request
from synapse import event_auth
from synapse.api.constants import Direction, EventTypes, Membership
from synapse.api.constants import Direction, EventTypes, Membership, StickyEvent
from synapse.api.errors import (
AuthError,
Codes,
@@ -210,6 +210,7 @@ class RoomStateEventRestServlet(RestServlet):
self.clock = hs.get_clock()
self._max_event_delay_ms = hs.config.server.max_event_delay_ms
self._spam_checker_module_callbacks = hs.get_module_api_callbacks().spam_checker
self._msc4354_enabled = hs.config.experimental.msc4354_enabled
def register(self, http_server: HttpServer) -> None:
# /rooms/$roomid/state/$eventtype
@@ -331,6 +332,10 @@ class RoomStateEventRestServlet(RestServlet):
if requester.app_service:
origin_server_ts = parse_integer(request, "ts")
sticky_duration_ms: int | None = None
if self._msc4354_enabled:
sticky_duration_ms = parse_integer(request, StickyEvent.QUERY_PARAM_NAME)
delay = _parse_request_delay(request, self._max_event_delay_ms)
if delay is not None:
delay_id = await self.delayed_events_handler.add(
@@ -341,6 +346,7 @@ class RoomStateEventRestServlet(RestServlet):
origin_server_ts=origin_server_ts,
content=content,
delay=delay,
sticky_duration_ms=sticky_duration_ms,
)
set_tag("delay_id", delay_id)
@@ -368,6 +374,10 @@ class RoomStateEventRestServlet(RestServlet):
"room_id": room_id,
"sender": requester.user.to_string(),
}
if sticky_duration_ms is not None:
event_dict[StickyEvent.FIELD_NAME] = {
"duration_ms": sticky_duration_ms,
}
if state_key is not None:
event_dict["state_key"] = state_key
@@ -400,6 +410,7 @@ class RoomSendEventRestServlet(TransactionRestServlet):
self.delayed_events_handler = hs.get_delayed_events_handler()
self.auth = hs.get_auth()
self._max_event_delay_ms = hs.config.server.max_event_delay_ms
self._msc4354_enabled = hs.config.experimental.msc4354_enabled
def register(self, http_server: HttpServer) -> None:
# /rooms/$roomid/send/$event_type[/$txn_id]
@@ -420,6 +431,10 @@ class RoomSendEventRestServlet(TransactionRestServlet):
if requester.app_service:
origin_server_ts = parse_integer(request, "ts")
sticky_duration_ms: int | None = None
if self._msc4354_enabled:
sticky_duration_ms = parse_integer(request, StickyEvent.QUERY_PARAM_NAME)
delay = _parse_request_delay(request, self._max_event_delay_ms)
if delay is not None:
delay_id = await self.delayed_events_handler.add(
@@ -430,6 +445,7 @@ class RoomSendEventRestServlet(TransactionRestServlet):
origin_server_ts=origin_server_ts,
content=content,
delay=delay,
sticky_duration_ms=sticky_duration_ms,
)
set_tag("delay_id", delay_id)
@@ -446,6 +462,11 @@ class RoomSendEventRestServlet(TransactionRestServlet):
if origin_server_ts is not None:
event_dict["origin_server_ts"] = origin_server_ts
if sticky_duration_ms is not None:
event_dict[StickyEvent.FIELD_NAME] = {
"duration_ms": sticky_duration_ms,
}
try:
(
event,
synapse/storage/databases/main/delayed_events.py
+8 -1
View File
@@ -54,6 +54,7 @@ class EventDetails:
origin_server_ts: Timestamp | None
content: JsonDict
device_id: DeviceID | None
sticky_duration_ms: int | None
@attr.s(slots=True, frozen=True, auto_attribs=True)
@@ -122,6 +123,7 @@ class DelayedEventsStore(SQLBaseStore):
origin_server_ts: int | None,
content: JsonDict,
delay: int,
sticky_duration_ms: int | None,
) -> tuple[DelayID, Timestamp]:
"""
Inserts a new delayed event in the DB.
@@ -148,6 +150,7 @@ class DelayedEventsStore(SQLBaseStore):
"state_key": state_key,
"origin_server_ts": origin_server_ts,
"content": json_encoder.encode(content),
"sticky_duration_ms": sticky_duration_ms,
},
)
@@ -299,6 +302,7 @@ class DelayedEventsStore(SQLBaseStore):
"send_ts",
"content",
"device_id",
"sticky_duration_ms",
)
)
sql_update = "UPDATE delayed_events SET is_processed = TRUE"
@@ -344,6 +348,7 @@ class DelayedEventsStore(SQLBaseStore):
Timestamp(row[5] if row[5] is not None else row[6]),
db_to_json(row[7]),
DeviceID(row[8]) if row[8] is not None else None,
int(row[9]) if row[9] is not None else None,
DelayID(row[0]),
UserLocalpart(row[1]),
)
@@ -392,6 +397,7 @@ class DelayedEventsStore(SQLBaseStore):
origin_server_ts,
content,
device_id,
sticky_duration_ms,
user_localpart
""",
(delay_id,),
@@ -407,8 +413,9 @@ class DelayedEventsStore(SQLBaseStore):
Timestamp(row[3]) if row[3] is not None else None,
db_to_json(row[4]),
DeviceID(row[5]) if row[5] is not None else None,
int(row[6]) if row[6] is not None else None,
DelayID(delay_id),
UserLocalpart(row[6]),
UserLocalpart(row[7]),
)
return event, self._get_next_delayed_event_send_ts_txn(txn)