mirror of
https://github.com/element-hq/synapse.git
synced 2026-11-04 11:17:51 +00:00
Trace the Rust sliding sync steps
Sliding sync requests computed in Rust now report spans under the request's Python span, one per step the handler calls: `sliding_sync.load_connection_state`, `sliding_sync.interested_rooms`, `sliding_sync.room_entries` (with the room count), and `sliding_sync.finish_request`. Inside the room entries, a `sliding_sync.room` per room is tagged `room_id` and its `outcome` (`entry`, `fallback` or `error`, marked with the `error` tag), with that room's reads under it. `sliding_sync.encode` covers encoding the rooms, which runs on a threadpool thread whose logcontext carries the request's Python span (`encode_json_response`). A trial test makes a sliding sync request over two rooms with tracing on and checks that each step is under the servlet's Python span, that each room is under `sliding_sync.room_entries`, and that the rooms' reads are under them. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01W4zxygfJC93SsugxskhoJG
This commit is contained in:
1 parent
a7ffaa582a
commit
1d812b46c2
4 files changed
+126
-2
No files matched your search
@@ -28,6 +28,7 @@ use crate::events::serialize::{
|
||||
serialize_event, BundleRef, SerializeEventConfig, SerializeLookups,
|
||||
};
|
||||
use crate::events::Event;
|
||||
use crate::logging::opentracing::layer::with_py_parent;
|
||||
use crate::sliding_sync::result::RoomResult;
|
||||
use crate::sliding_sync::room_entry::RoomEntry;
|
||||
use crate::sync::relations::Bundle;
|
||||
@@ -183,8 +184,13 @@ pub fn encode_rooms<'py>(
|
||||
) -> PyResult<Bound<'py, PyBytes>> {
|
||||
let unsigned_additions = unsigned_additions.unwrap_or_default();
|
||||
let entries: Vec<Arc<RoomEntry>> = rooms.iter().map(|room| room.get().entry.clone()).collect();
|
||||
let encoded =
|
||||
py.detach(|| encode_rooms_json(&entries, time_now_ms, config, &unsigned_additions))?;
|
||||
// On a threadpool thread (`defer_to_thread`), whose logcontext carries
|
||||
// the request's Python span.
|
||||
let encoded = with_py_parent(py, || {
|
||||
let _span =
|
||||
tracing::info_span!("sliding_sync.encode", rooms = entries.len() as u64).entered();
|
||||
py.detach(|| encode_rooms_json(&entries, time_now_ms, config, &unsigned_additions))
|
||||
})?;
|
||||
Ok(PyBytes::new(py, &encoded))
|
||||
}
|
||||
|
||||
|
||||
@@ -45,6 +45,7 @@ use room_lists::filter::NotImplementedFilter;
|
||||
use room_lists::interested::{InterestedRooms, ListsSnapshot, PyInterestedRooms};
|
||||
use room_lists::lists::UnknownPosition;
|
||||
use state::{ConnectionState, ConnectionStateData, InstanceMap};
|
||||
use tracing::Instrument;
|
||||
|
||||
use crate::caches::lock_ignoring_poison;
|
||||
use crate::database::postgres::pool::{ConnectionPool, PyConnectionPool};
|
||||
@@ -213,6 +214,7 @@ impl SlidingSyncHandler {
|
||||
let stats = SlidingSyncStats::from_db(&reader.stats());
|
||||
Ok::<_, anyhow::Error>(ConnectionState::new(data, instances, stats))
|
||||
}
|
||||
.instrument(tracing::info_span!("sliding_sync.load_connection_state"))
|
||||
.await;
|
||||
result.map_err(error_to_py)
|
||||
}))
|
||||
@@ -236,6 +238,7 @@ impl SlidingSyncHandler {
|
||||
let interested = room_lists::interested::compute_interested_rooms(
|
||||
&reader, &request, &previous, snapshot, now_ms,
|
||||
)
|
||||
.instrument(tracing::info_span!("sliding_sync.interested_rooms"))
|
||||
.await
|
||||
.map_err(error_to_py)?;
|
||||
Ok(PyInterestedRooms {
|
||||
@@ -272,8 +275,10 @@ impl SlidingSyncHandler {
|
||||
let now_ms = self.runtime.clock().now_millis() as i64;
|
||||
Ok(Box::pin(async move {
|
||||
let reader = store.sliding_sync_reader(pool);
|
||||
let span = tracing::info_span!("sliding_sync.room_entries", rooms = rooms.len() as u64);
|
||||
let outcomes =
|
||||
room_entry::generate_room_entries(&reader, &request, &instance_ids, now_ms, &rooms)
|
||||
.instrument(span)
|
||||
.await
|
||||
.map_err(anyhow_to_py)?;
|
||||
let stats = SlidingSyncStats::from_db(&reader.stats());
|
||||
@@ -310,6 +315,7 @@ impl SlidingSyncHandler {
|
||||
&interested,
|
||||
now_ms,
|
||||
)
|
||||
.instrument(tracing::info_span!("sliding_sync.finish_request"))
|
||||
.await
|
||||
.map_err(error_to_py)?;
|
||||
Ok((position, SlidingSyncStats::from_db(&reader.stats())))
|
||||
|
||||
@@ -28,8 +28,10 @@ use anyhow::{bail, Context};
|
||||
use futures::{stream, FutureExt, StreamExt};
|
||||
use indexmap::IndexMap;
|
||||
use serde_json::{Map, Value};
|
||||
use tracing::Instrument;
|
||||
|
||||
use crate::events::Event;
|
||||
use crate::logging::opentracing::record_error;
|
||||
use crate::sliding_sync::config::{RoomSyncConfig, MEMBER};
|
||||
use crate::sliding_sync::connection::{HaveSentRoom, RoomLazyMembershipChanges};
|
||||
use crate::sliding_sync::request::SlidingSyncRequestData;
|
||||
@@ -877,8 +879,15 @@ async fn room_outcome(
|
||||
context: &RoomContext<'_, '_>,
|
||||
input: &RoomInputData,
|
||||
) -> (Arc<str>, RoomOutcome) {
|
||||
let span = tracing::info_span!(
|
||||
"sliding_sync.room",
|
||||
room_id = &*input.membership.room_id,
|
||||
outcome = tracing::field::Empty,
|
||||
error = tracing::field::Empty,
|
||||
);
|
||||
let outcome = match AssertUnwindSafe(room_sync_data(context, input))
|
||||
.catch_unwind()
|
||||
.instrument(span.clone())
|
||||
.await
|
||||
{
|
||||
Ok(Ok(computed)) => RoomOutcome::Computed(Box::new(computed)),
|
||||
@@ -888,6 +897,15 @@ async fn room_outcome(
|
||||
},
|
||||
Err(panic) => RoomOutcome::Error(format!("panicked: {}", panic_message(&*panic))),
|
||||
};
|
||||
let outcome_name = match &outcome {
|
||||
RoomOutcome::Computed(_) => "entry",
|
||||
RoomOutcome::Fallback(_) => "fallback",
|
||||
RoomOutcome::Error(_) => "error",
|
||||
};
|
||||
span.record("outcome", outcome_name);
|
||||
if let RoomOutcome::Error(error) = &outcome {
|
||||
record_error(&span, error);
|
||||
}
|
||||
(Arc::clone(&input.membership.room_id), outcome)
|
||||
}
|
||||
|
||||
|
||||
@@ -738,3 +738,97 @@ class EventLoadTracingTestCase(_AgentTestCase):
|
||||
for s in spans
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class SlidingSyncTracingTestCase(_AgentTestCase):
|
||||
"""A sliding sync request Rust computes reports its steps and a span per
|
||||
room under the request's Python span.
|
||||
|
||||
Needs the native Rust Postgres driver:
|
||||
|
||||
SYNAPSE_TEST_RUST_DRIVER=1 POSTGRES=1 run-trial.sh tests.logging.test_rust_tracing
|
||||
"""
|
||||
|
||||
servlets = [
|
||||
admin.register_servlets,
|
||||
room.register_servlets,
|
||||
login.register_servlets,
|
||||
sync.register_servlets,
|
||||
]
|
||||
|
||||
def default_config(self) -> dict[str, Any]:
|
||||
config = super().default_config()
|
||||
config.setdefault("experimental_features", {})["rust_sliding_sync"] = "on"
|
||||
return config
|
||||
|
||||
def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
|
||||
hs = self.setup_test_homeserver()
|
||||
if hs.get_sliding_sync_handler()._rust_sliding_sync is None:
|
||||
raise SkipTest("needs the Rust sliding sync handler")
|
||||
return hs
|
||||
|
||||
def prepare_before_tracing(self) -> None:
|
||||
self.user = self.register_user("user", "pass")
|
||||
self.token = self.login("user", "pass")
|
||||
self.rooms = [
|
||||
self.helper.create_room_as(self.user, tok=self.token) for _ in range(2)
|
||||
]
|
||||
|
||||
def test_sliding_sync_span_tree(self) -> None:
|
||||
channel = self.make_request(
|
||||
"POST",
|
||||
"/_matrix/client/unstable/org.matrix.simplified_msc3575/sync",
|
||||
{
|
||||
"lists": {
|
||||
"all": {
|
||||
"ranges": [[0, 9]],
|
||||
"required_state": [],
|
||||
"timeline_limit": 1,
|
||||
}
|
||||
}
|
||||
},
|
||||
access_token=self.token,
|
||||
)
|
||||
self.assertEqual(channel.code, 200, channel.json_body)
|
||||
|
||||
spans = self._received_spans()
|
||||
python_by_id = self._python_spans_by_id()
|
||||
|
||||
def named(name: str) -> "list[ttypes.Span]":
|
||||
return [s for s in spans if s.operationName == name]
|
||||
|
||||
steps = [
|
||||
"sliding_sync.load_connection_state",
|
||||
"sliding_sync.interested_rooms",
|
||||
"sliding_sync.room_entries",
|
||||
"sliding_sync.finish_request",
|
||||
"sliding_sync.encode",
|
||||
]
|
||||
for step in steps:
|
||||
(span,) = named(step)
|
||||
self.assertIn(
|
||||
"SlidingSyncRestServlet", self._python_ancestors(span, python_by_id)
|
||||
)
|
||||
|
||||
(entries,) = named("sliding_sync.room_entries")
|
||||
rooms = named("sliding_sync.room")
|
||||
self.assertEqual(len(rooms), 2)
|
||||
self.assertEqual(
|
||||
{next(t.vStr for t in s.tags if t.key == "room_id") for s in rooms},
|
||||
set(self.rooms),
|
||||
)
|
||||
for room_span in rooms:
|
||||
self.assertEqual(room_span.parentSpanId, entries.spanId)
|
||||
self.assertEqual(
|
||||
next(t.vStr for t in room_span.tags if t.key == "outcome"), "entry"
|
||||
)
|
||||
|
||||
# The rooms' reads nest under them.
|
||||
room_span_ids = {room_span.spanId for room_span in rooms}
|
||||
self.assertTrue(
|
||||
any(
|
||||
s.operationName == "db.connection" and s.parentSpanId in room_span_ids
|
||||
for s in spans
|
||||
),
|
||||
[s.operationName for s in spans],
|
||||
)
|
||||
Reference in new issue
Block a user