|
43 | 43 | from synapse.storage.database import DatabasePool
|
44 | 44 | from synapse.storage.util.id_generators import StreamIdGenerator
|
45 | 45 | from synapse.types import get_domain_from_id
|
46 |
| -from synapse.util.caches.descriptors import Cache, cached, cachedInlineCallbacks |
| 46 | +from synapse.util.caches.descriptors import Cache, cachedInlineCallbacks |
47 | 47 | from synapse.util.iterutils import batch_iter
|
48 | 48 | from synapse.util.metrics import Measure
|
49 | 49 |
|
@@ -137,42 +137,6 @@ def get_received_ts(self, event_id):
|
137 | 137 | desc="get_received_ts",
|
138 | 138 | )
|
139 | 139 |
|
140 |
| - def get_received_ts_by_stream_pos(self, stream_ordering): |
141 |
| - """Given a stream ordering get an approximate timestamp of when it |
142 |
| - happened. |
143 |
| -
|
144 |
| - This is done by simply taking the received ts of the first event that |
145 |
| - has a stream ordering greater than or equal to the given stream pos. |
146 |
| - If none exists returns the current time, on the assumption that it must |
147 |
| - have happened recently. |
148 |
| -
|
149 |
| - Args: |
150 |
| - stream_ordering (int) |
151 |
| -
|
152 |
| - Returns: |
153 |
| - Deferred[int] |
154 |
| - """ |
155 |
| - |
156 |
| - def _get_approximate_received_ts_txn(txn): |
157 |
| - sql = """ |
158 |
| - SELECT received_ts FROM events |
159 |
| - WHERE stream_ordering >= ? |
160 |
| - LIMIT 1 |
161 |
| - """ |
162 |
| - |
163 |
| - txn.execute(sql, (stream_ordering,)) |
164 |
| - row = txn.fetchone() |
165 |
| - if row and row[0]: |
166 |
| - ts = row[0] |
167 |
| - else: |
168 |
| - ts = self.clock.time_msec() |
169 |
| - |
170 |
| - return ts |
171 |
| - |
172 |
| - return self.db_pool.runInteraction( |
173 |
| - "get_approximate_received_ts", _get_approximate_received_ts_txn |
174 |
| - ) |
175 |
| - |
176 | 140 | @defer.inlineCallbacks
|
177 | 141 | def get_event(
|
178 | 142 | self,
|
@@ -923,36 +887,6 @@ def have_seen_events_txn(txn, chunk):
|
923 | 887 | )
|
924 | 888 | return results
|
925 | 889 |
|
926 |
| - def _get_total_state_event_counts_txn(self, txn, room_id): |
927 |
| - """ |
928 |
| - See get_total_state_event_counts. |
929 |
| - """ |
930 |
| - # We join against the events table as that has an index on room_id |
931 |
| - sql = """ |
932 |
| - SELECT COUNT(*) FROM state_events |
933 |
| - INNER JOIN events USING (room_id, event_id) |
934 |
| - WHERE room_id=? |
935 |
| - """ |
936 |
| - txn.execute(sql, (room_id,)) |
937 |
| - row = txn.fetchone() |
938 |
| - return row[0] if row else 0 |
939 |
| - |
940 |
| - def get_total_state_event_counts(self, room_id): |
941 |
| - """ |
942 |
| - Gets the total number of state events in a room. |
943 |
| -
|
944 |
| - Args: |
945 |
| - room_id (str) |
946 |
| -
|
947 |
| - Returns: |
948 |
| - Deferred[int] |
949 |
| - """ |
950 |
| - return self.db_pool.runInteraction( |
951 |
| - "get_total_state_event_counts", |
952 |
| - self._get_total_state_event_counts_txn, |
953 |
| - room_id, |
954 |
| - ) |
955 |
| - |
956 | 890 | def _get_current_state_event_counts_txn(self, txn, room_id):
|
957 | 891 | """
|
958 | 892 | See get_current_state_event_counts.
|
@@ -1222,97 +1156,6 @@ def get_deltas_for_stream_id_txn(txn, stream_id):
|
1222 | 1156 |
|
1223 | 1157 | return rows, to_token, True
|
1224 | 1158 |
|
1225 |
| - @cached(num_args=5, max_entries=10) |
1226 |
| - def get_all_new_events( |
1227 |
| - self, |
1228 |
| - last_backfill_id, |
1229 |
| - last_forward_id, |
1230 |
| - current_backfill_id, |
1231 |
| - current_forward_id, |
1232 |
| - limit, |
1233 |
| - ): |
1234 |
| - """Get all the new events that have arrived at the server either as |
1235 |
| - new events or as backfilled events""" |
1236 |
| - have_backfill_events = last_backfill_id != current_backfill_id |
1237 |
| - have_forward_events = last_forward_id != current_forward_id |
1238 |
| - |
1239 |
| - if not have_backfill_events and not have_forward_events: |
1240 |
| - return defer.succeed(AllNewEventsResult([], [], [], [], [])) |
1241 |
| - |
1242 |
| - def get_all_new_events_txn(txn): |
1243 |
| - sql = ( |
1244 |
| - "SELECT e.stream_ordering, e.event_id, e.room_id, e.type," |
1245 |
| - " state_key, redacts" |
1246 |
| - " FROM events AS e" |
1247 |
| - " LEFT JOIN redactions USING (event_id)" |
1248 |
| - " LEFT JOIN state_events USING (event_id)" |
1249 |
| - " WHERE ? < stream_ordering AND stream_ordering <= ?" |
1250 |
| - " ORDER BY stream_ordering ASC" |
1251 |
| - " LIMIT ?" |
1252 |
| - ) |
1253 |
| - if have_forward_events: |
1254 |
| - txn.execute(sql, (last_forward_id, current_forward_id, limit)) |
1255 |
| - new_forward_events = txn.fetchall() |
1256 |
| - |
1257 |
| - if len(new_forward_events) == limit: |
1258 |
| - upper_bound = new_forward_events[-1][0] |
1259 |
| - else: |
1260 |
| - upper_bound = current_forward_id |
1261 |
| - |
1262 |
| - sql = ( |
1263 |
| - "SELECT event_stream_ordering, event_id, state_group" |
1264 |
| - " FROM ex_outlier_stream" |
1265 |
| - " WHERE ? > event_stream_ordering" |
1266 |
| - " AND event_stream_ordering >= ?" |
1267 |
| - " ORDER BY event_stream_ordering DESC" |
1268 |
| - ) |
1269 |
| - txn.execute(sql, (last_forward_id, upper_bound)) |
1270 |
| - forward_ex_outliers = txn.fetchall() |
1271 |
| - else: |
1272 |
| - new_forward_events = [] |
1273 |
| - forward_ex_outliers = [] |
1274 |
| - |
1275 |
| - sql = ( |
1276 |
| - "SELECT -e.stream_ordering, e.event_id, e.room_id, e.type," |
1277 |
| - " state_key, redacts" |
1278 |
| - " FROM events AS e" |
1279 |
| - " LEFT JOIN redactions USING (event_id)" |
1280 |
| - " LEFT JOIN state_events USING (event_id)" |
1281 |
| - " WHERE ? > stream_ordering AND stream_ordering >= ?" |
1282 |
| - " ORDER BY stream_ordering DESC" |
1283 |
| - " LIMIT ?" |
1284 |
| - ) |
1285 |
| - if have_backfill_events: |
1286 |
| - txn.execute(sql, (-last_backfill_id, -current_backfill_id, limit)) |
1287 |
| - new_backfill_events = txn.fetchall() |
1288 |
| - |
1289 |
| - if len(new_backfill_events) == limit: |
1290 |
| - upper_bound = new_backfill_events[-1][0] |
1291 |
| - else: |
1292 |
| - upper_bound = current_backfill_id |
1293 |
| - |
1294 |
| - sql = ( |
1295 |
| - "SELECT -event_stream_ordering, event_id, state_group" |
1296 |
| - " FROM ex_outlier_stream" |
1297 |
| - " WHERE ? > event_stream_ordering" |
1298 |
| - " AND event_stream_ordering >= ?" |
1299 |
| - " ORDER BY event_stream_ordering DESC" |
1300 |
| - ) |
1301 |
| - txn.execute(sql, (-last_backfill_id, -upper_bound)) |
1302 |
| - backward_ex_outliers = txn.fetchall() |
1303 |
| - else: |
1304 |
| - new_backfill_events = [] |
1305 |
| - backward_ex_outliers = [] |
1306 |
| - |
1307 |
| - return AllNewEventsResult( |
1308 |
| - new_forward_events, |
1309 |
| - new_backfill_events, |
1310 |
| - forward_ex_outliers, |
1311 |
| - backward_ex_outliers, |
1312 |
| - ) |
1313 |
| - |
1314 |
| - return self.db_pool.runInteraction("get_all_new_events", get_all_new_events_txn) |
1315 |
| - |
1316 | 1159 | async def is_event_after(self, event_id1, event_id2):
|
1317 | 1160 | """Returns True if event_id1 is after event_id2 in the stream
|
1318 | 1161 | """
|
@@ -1357,14 +1200,3 @@ def get_next_event_to_expire_txn(txn):
|
1357 | 1200 | return self.db_pool.runInteraction(
|
1358 | 1201 | desc="get_next_event_to_expire", func=get_next_event_to_expire_txn
|
1359 | 1202 | )
|
1360 |
| - |
1361 |
| - |
1362 |
| -AllNewEventsResult = namedtuple( |
1363 |
| - "AllNewEventsResult", |
1364 |
| - [ |
1365 |
| - "new_forward_events", |
1366 |
| - "new_backfill_events", |
1367 |
| - "forward_ex_outliers", |
1368 |
| - "backward_ex_outliers", |
1369 |
| - ], |
1370 |
| -) |
0 commit comments