Skip to content

Commit 0744d75

Browse files
michellebriermichellebrier
authored
add trusted notifier discrepancies to health check + index and delist sequentially (#5565)
1 parent dacfa71 commit 0744d75

File tree

6 files changed

+130
-26
lines changed

6 files changed

+130
-26
lines changed

‎discovery-provider/src/app.py‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -413,10 +413,6 @@ def configure_celery(celery, test_config=None):
413413
"task": "index_reactions",
414414
"schedule": timedelta(seconds=5),
415415
},
416-
"update_delist_statuses": {
417-
"task": "update_delist_statuses",
418-
"schedule": timedelta(seconds=20),
419-
},
420416
"index_profile_challenge_backfill": {
421417
"task": "index_profile_challenge_backfill",
422418
"schedule": timedelta(minutes=1),

‎discovery-provider/src/queries/get_health.py‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,12 @@
1-
import json
21
import logging
32
import os
43
import time
5-
from datetime import datetime, timedelta, timezone
4+
from datetime import datetime
65
from typing import Dict, Optional, Tuple, TypedDict, cast
76

87
import requests
98
from elasticsearch import Elasticsearch
109
from redis import Redis
11-
from sqlalchemy.sql import text
1210
from src.eth_indexing.event_scanner import eth_indexing_last_scanned_block_key
1311
from src.models.indexing.block import Block
1412
from src.monitors import monitor_names, monitors
@@ -22,6 +20,7 @@
2220
from src.queries.get_sol_rewards_manager import get_sol_rewards_manager_health_info
2321
from src.queries.get_sol_user_bank import get_sol_user_bank_health_info
2422
from src.queries.get_spl_audio import get_spl_audio_health_info
23+
from src.queries.get_trusted_notifier_discrepancies import get_delist_statuses_ok
2524
from src.utils import (
2625
db_session,
2726
get_all_other_nodes,
@@ -296,6 +295,7 @@ def get_health(args: GetHealthArgs, use_redis_cache: bool = True) -> Tuple[Dict,
296295
content_nodes = get_all_other_nodes.get_all_healthy_content_nodes_cached(redis)
297296
final_poa_block = helpers.get_final_poa_block()
298297
backfilled_cid_data = get_backfilled_cid_data(redis)
298+
delist_statuses_ok = get_delist_statuses_ok()
299299
health_results = {
300300
"web": {
301301
"blocknumber": latest_block_num,
@@ -332,6 +332,7 @@ def get_health(args: GetHealthArgs, use_redis_cache: bool = True) -> Tuple[Dict,
332332
"final_poa_block": final_poa_block,
333333
"network": {"discovery_nodes": discovery_nodes, "content_nodes": content_nodes},
334334
"backfilled_cid_data": backfilled_cid_data,
335+
"delist_statuses_ok": delist_statuses_ok,
335336
}
336337

337338
if os.getenv("AUDIUS_DOCKER_COMPOSE_GIT_SHA") is not None:
@@ -411,6 +412,7 @@ def get_health(args: GetHealthArgs, use_redis_cache: bool = True) -> Tuple[Dict,
411412
or unhealthy_challenges
412413
or play_health_info["is_unhealthy"]
413414
or reactions_health_info["is_unhealthy"]
415+
or not delist_statuses_ok
414416
)
415417

416418
return health_results, is_unhealthy

‎discovery-provider/src/queries/get_health_unit_test.py‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import json
22
import os
3+
from datetime import datetime, timezone
34
from time import time
45
from unittest.mock import MagicMock, patch
56

@@ -8,6 +9,14 @@
89
from src.models.indexing.indexing_checkpoints import IndexingCheckpoint
910
from src.queries.get_health import get_health
1011
from src.utils.redis_constants import (
12+
TRACK_DELIST_DISCREPANCIES_KEY,
13+
TRACK_DELIST_DISCREPANCIES_TIMESTAMP_KEY,
14+
TRACK_DELIST_STATUS_CURSOR_CHECK_KEY,
15+
TRACK_DELIST_STATUS_CURSOR_CHECK_TIMESTAMP_KEY,
16+
USER_DELIST_DISCREPANCIES_KEY,
17+
USER_DELIST_DISCREPANCIES_TIMESTAMP_KEY,
18+
USER_DELIST_STATUS_CURSOR_CHECK_KEY,
19+
USER_DELIST_STATUS_CURSOR_CHECK_TIMESTAMP_KEY,
1120
challenges_last_processed_event_redis_key,
1221
latest_block_hash_redis_key,
1322
latest_block_redis_key,
@@ -124,6 +133,29 @@ def cache_play_health_vars(redis_mock):
124133
)
125134

126135

136+
def cache_trusted_notifier_discrepancies_vars(redis_mock):
137+
redis_mock.set(
138+
USER_DELIST_STATUS_CURSOR_CHECK_TIMESTAMP_KEY,
139+
datetime.now(timezone.utc).timestamp(),
140+
)
141+
redis_mock.set(USER_DELIST_STATUS_CURSOR_CHECK_KEY, "ok")
142+
redis_mock.set(
143+
TRACK_DELIST_STATUS_CURSOR_CHECK_TIMESTAMP_KEY,
144+
datetime.now(timezone.utc).timestamp(),
145+
)
146+
redis_mock.set(TRACK_DELIST_STATUS_CURSOR_CHECK_KEY, "ok")
147+
redis_mock.set(
148+
USER_DELIST_DISCREPANCIES_TIMESTAMP_KEY,
149+
datetime.now(timezone.utc).timestamp(),
150+
)
151+
redis_mock.set(USER_DELIST_DISCREPANCIES_KEY, "[]")
152+
redis_mock.set(
153+
TRACK_DELIST_DISCREPANCIES_TIMESTAMP_KEY,
154+
datetime.now(timezone.utc).timestamp(),
155+
)
156+
redis_mock.set(TRACK_DELIST_DISCREPANCIES_KEY, "[]")
157+
158+
127159
def test_get_health(web3_mock, redis_mock, db_mock):
128160
"""Tests that the health check returns db data"""
129161

@@ -135,6 +167,7 @@ def get_block(_u1, _u2): # unused
135167
return block
136168

137169
cache_play_health_vars(redis_mock)
170+
cache_trusted_notifier_discrepancies_vars(redis_mock)
138171
web3_mock.eth.get_block = get_block
139172

140173
# Set up db state
@@ -177,6 +210,7 @@ def get_block(_u1, _u2): # unused
177210
return block
178211

179212
cache_play_health_vars(redis_mock)
213+
cache_trusted_notifier_discrepancies_vars(redis_mock)
180214
web3_mock.eth.get_block = get_block
181215

182216
# Set up redis state
@@ -224,6 +258,7 @@ def get_block(_u1, _u2): # unused
224258
return block
225259

226260
cache_play_health_vars(redis_mock)
261+
cache_trusted_notifier_discrepancies_vars(redis_mock)
227262
web3_mock.eth.get_block = get_block
228263

229264
# Set up redis state
@@ -269,6 +304,7 @@ def get_block(_u1, _u2): # unused
269304
return block
270305

271306
cache_play_health_vars(redis_mock)
307+
cache_trusted_notifier_discrepancies_vars(redis_mock)
272308
web3_mock.eth.get_block = get_block
273309

274310
# Set up db state
@@ -310,6 +346,7 @@ def get_block(_u1, _u2): # unused
310346
return block
311347

312348
cache_play_health_vars(redis_mock)
349+
cache_trusted_notifier_discrepancies_vars(redis_mock)
313350
web3_mock.eth.get_block = get_block
314351

315352
# Set up redis state
@@ -358,6 +395,7 @@ def get_block(_u1, _u2): # unused
358395
return block
359396

360397
cache_play_health_vars(redis_mock)
398+
cache_trusted_notifier_discrepancies_vars(redis_mock)
361399
web3_mock.eth.get_block = get_block
362400

363401
# Set up db state
@@ -395,6 +433,7 @@ def get_block(_u1, _u2): # unused
395433
return block
396434

397435
cache_play_health_vars(redis_mock)
436+
cache_trusted_notifier_discrepancies_vars(redis_mock)
398437
web3_mock.eth.get_block = get_block
399438

400439
# Set up db state
@@ -446,6 +485,7 @@ def get_block(_u1, _u2): # unused
446485

447486
web3_mock.eth.get_block = get_block
448487
cache_play_health_vars(redis_mock)
488+
cache_trusted_notifier_discrepancies_vars(redis_mock)
449489

450490
# Set up db state
451491
with db_mock.scoped_session() as session:
@@ -501,6 +541,7 @@ def get_block(_u1, _u2): # unused
501541
return block
502542

503543
cache_play_health_vars(redis_mock)
544+
cache_trusted_notifier_discrepancies_vars(redis_mock)
504545
web3_mock.eth.get_block = get_block
505546

506547
# Set up db state
@@ -556,6 +597,7 @@ def get_block(_u1, _u2): # unused
556597
return block
557598

558599
cache_play_health_vars(redis_mock)
600+
cache_trusted_notifier_discrepancies_vars(redis_mock)
559601
web3_mock.eth.get_block = get_block
560602

561603
# Set up redis state

‎discovery-provider/src/queries/get_trusted_notifier_discrepancies.py‎

Lines changed: 34 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,13 @@
11
import json
2-
import logging
32
from datetime import datetime, timedelta, timezone
43

54
from redis import Redis
6-
from sqlalchemy import desc, func
5+
from sqlalchemy import desc
76
from sqlalchemy.sql import text
87
from src.models.delisting.delist_status_cursor import DelistEntity, DelistStatusCursor
98
from src.models.delisting.track_delist_status import TrackDelistStatus
109
from src.models.delisting.user_delist_status import UserDelistStatus
11-
from src.utils import db_session, helpers, redis_connection
10+
from src.utils import db_session, redis_connection
1211
from src.utils.redis_constants import (
1312
TRACK_DELIST_DISCREPANCIES_KEY,
1413
TRACK_DELIST_DISCREPANCIES_TIMESTAMP_KEY,
@@ -19,6 +18,9 @@
1918
USER_DELIST_STATUS_CURSOR_CHECK_KEY,
2019
USER_DELIST_STATUS_CURSOR_CHECK_TIMESTAMP_KEY,
2120
)
21+
from src.utils.structured_logger import StructuredLogger
22+
23+
logger = StructuredLogger(__name__)
2224

2325

2426
def check_user_delist_status_cursor(redis: Redis):
@@ -67,7 +69,10 @@ def check_user_delist_status_cursor(redis: Redis):
6769
redis.set(USER_DELIST_STATUS_CURSOR_CHECK_KEY, ok)
6870
return ok
6971
except Exception as e:
70-
logging.error("issue with user delist discrepancies %s", exc_info=e)
72+
logger.error(
73+
"get_trusted_notifier_discrepancies.py | issue with user delist cursor check %s",
74+
exc_info=e,
75+
)
7176
pass
7277

7378

@@ -117,7 +122,10 @@ def check_track_delist_status_cursor(redis: Redis):
117122
redis.set(TRACK_DELIST_STATUS_CURSOR_CHECK_KEY, ok)
118123
return ok
119124
except Exception as e:
120-
logging.error("issue with track delist discrepancies %s", exc_info=e)
125+
logger.error(
126+
"get_trusted_notifier_discrepancies.py | issue with track delist cursor check %s",
127+
exc_info=e,
128+
)
121129
pass
122130

123131

@@ -171,14 +179,21 @@ def get_user_delist_discrepancies(redis: Redis):
171179
user_delist_discrepancies = json.dumps(
172180
[dict(row) for row in result], default=str
173181
)
182+
if user_delist_discrepancies != "[]":
183+
logger.info(
184+
f"get_trusted_notifier_discrepancies.py | found user delist discrepancies: {user_delist_discrepancies}"
185+
)
174186
redis.set(
175187
USER_DELIST_DISCREPANCIES_TIMESTAMP_KEY,
176188
datetime.now(timezone.utc).timestamp(),
177189
)
178190
redis.set(USER_DELIST_DISCREPANCIES_KEY, user_delist_discrepancies)
179191
return user_delist_discrepancies
180192
except Exception as e:
181-
logging.error("issue with user delist discrepancies %s", exc_info=e)
193+
logger.error(
194+
"get_trusted_notifier_discrepancies.py | issue with user delist discrepancies %s",
195+
exc_info=e,
196+
)
182197
pass
183198

184199

@@ -232,14 +247,21 @@ def get_track_delist_discrepancies(redis: Redis):
232247
track_delist_discrepancies = json.dumps(
233248
[dict(row) for row in result], default=str
234249
)
250+
if track_delist_discrepancies != "[]":
251+
logger.info(
252+
f"get_trusted_notifier_discrepancies.py | found track delist discrepancies: {track_delist_discrepancies}"
253+
)
235254
redis.set(
236255
TRACK_DELIST_DISCREPANCIES_TIMESTAMP_KEY,
237256
datetime.now(timezone.utc).timestamp(),
238257
)
239258
redis.set(TRACK_DELIST_DISCREPANCIES_KEY, track_delist_discrepancies)
240259
return track_delist_discrepancies
241260
except Exception as e:
242-
logging.error("issue with track delist discrepancies %s", exc_info=e)
261+
logger.error(
262+
"get_trusted_notifier_discrepancies.py | issue with track delist discrepancies %s",
263+
exc_info=e,
264+
)
243265
pass
244266

245267

@@ -267,3 +289,8 @@ def get_trusted_notifier_discrepancies():
267289
or track_delist_discrepancies != "[]"
268290
)
269291
return health_results, is_unhealthy
292+
293+
294+
def get_delist_statuses_ok():
295+
_, is_unhealthy = get_trusted_notifier_discrepancies()
296+
return not is_unhealthy

‎discovery-provider/src/tasks/index_nethermind.py‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -444,6 +444,18 @@ def index_next_block(
444444
f"Error in calling update trending challenge {e}",
445445
exc_info=True,
446446
)
447+
try:
448+
# Every 100 blocks, poll and apply delist statuses from trusted notifier
449+
if next_block.number % 100 == 0:
450+
celery.send_task(
451+
"update_delist_statuses"
452+
)
453+
except Exception as e:
454+
# Do not throw error, as this should not stop indexing
455+
logger.error(
456+
f"Error in calling update_delist_statuses {e}",
457+
exc_info=True,
458+
)
447459
if skip_tx_hash:
448460
clear_indexing_error(redis)
449461

0 commit comments

Comments
 (0)