Fix race condition in room stats. (#6029)
Broke in #5971 Basically the bug is that if get_current_state_deltas returns no new updates and we then take the max pos, its possible that we miss an update that happens in between the two calls. (e.g. get_current_state_deltas looks up to stream pos 5, then an event persists and so getting the max stream pos returns 6, meaning that next time we check for things with a stream pos bigger than 6)
This commit is contained in:
parent
1e19ce00bf
commit
70c52821ce
|
@ -0,0 +1 @@
|
||||||
|
Fix room and user stats tracking.
|
|
@ -84,6 +84,13 @@ class StatsHandler(StateDeltasHandler):
|
||||||
# Loop round handling deltas until we're up to date
|
# Loop round handling deltas until we're up to date
|
||||||
|
|
||||||
while True:
|
while True:
|
||||||
|
# Be sure to read the max stream_ordering *before* checking if there are any outstanding
|
||||||
|
# deltas, since there is otherwise a chance that we could miss updates which arrive
|
||||||
|
# after we check the deltas.
|
||||||
|
room_max_stream_ordering = yield self.store.get_room_max_stream_ordering()
|
||||||
|
if self.pos == room_max_stream_ordering:
|
||||||
|
break
|
||||||
|
|
||||||
deltas = yield self.store.get_current_state_deltas(self.pos)
|
deltas = yield self.store.get_current_state_deltas(self.pos)
|
||||||
|
|
||||||
if deltas:
|
if deltas:
|
||||||
|
@ -94,7 +101,7 @@ class StatsHandler(StateDeltasHandler):
|
||||||
else:
|
else:
|
||||||
room_deltas = {}
|
room_deltas = {}
|
||||||
user_deltas = {}
|
user_deltas = {}
|
||||||
max_pos = yield self.store.get_room_max_stream_ordering()
|
max_pos = room_max_stream_ordering
|
||||||
|
|
||||||
# Then count deltas for total_events and total_event_bytes.
|
# Then count deltas for total_events and total_event_bytes.
|
||||||
room_count, user_count = yield self.store.get_changes_room_total_events_and_bytes(
|
room_count, user_count = yield self.store.get_changes_room_total_events_and_bytes(
|
||||||
|
@ -117,10 +124,9 @@ class StatsHandler(StateDeltasHandler):
|
||||||
stream_id=max_pos,
|
stream_id=max_pos,
|
||||||
)
|
)
|
||||||
|
|
||||||
event_processing_positions.labels("stats").set(max_pos)
|
logger.debug("Handled room stats to %s -> %s", self.pos, max_pos)
|
||||||
|
|
||||||
if self.pos == max_pos:
|
event_processing_positions.labels("stats").set(max_pos)
|
||||||
break
|
|
||||||
|
|
||||||
self.pos = max_pos
|
self.pos = max_pos
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue