Skip to content

Commit 0c98e92

Browse files
eclipsevortexeclipsevortex
committed
quick fix
1 parent a383205 commit 0c98e92

1 file changed

Lines changed: 133 additions & 49 deletions

File tree

‎subvortex/core/metagraph/metagraph.py‎

Lines changed: 133 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -198,21 +198,33 @@ async def start(self):
198198
)
199199

200200
# Sync from chain and update Redis
201-
axons, has_missing_country = await self._resync(
202-
last_update=last_update
203-
)
201+
try:
202+
axons, has_missing_country = await self._resync(
203+
last_update=last_update
204+
)
204205

205-
# Store the sync block
206-
last_synced_block = block
206+
# Store the sync block
207+
last_synced_block = block
207208

208-
# Notify listener the metagraph is ready with retry logic
209-
await self._notify_with_retry(state == "ready")
209+
# Notify listener the metagraph is ready with retry logic
210+
await self._notify_with_retry(state == "ready")
211+
212+
except Exception as resync_error:
213+
btul.logging.error(
214+
f"❌ Resync failed: {resync_error}",
215+
prefix=self.settings.logging_name,
216+
)
217+
# Mark metagraph as unready due to resync failure
218+
await self._mark_unready_on_error(f"Resync failed: {resync_error}")
219+
# Continue loop to retry
210220

211221
# Store the new axons
212222
axons = new_axons
213223

214224
except ConnectionRefusedError as e:
215225
btul.logging.error(f"Connection refused: {e}")
226+
# Mark metagraph as unready due to connection issues
227+
await self._mark_unready_on_error("Connection refused")
216228
await asyncio.sleep(1)
217229

218230
except Exception as e:
@@ -223,6 +235,8 @@ async def start(self):
223235
btul.logging.debug(
224236
traceback.format_exc(), prefix=self.settings.logging_name
225237
)
238+
# Mark metagraph as unready due to unhandled error
239+
await self._mark_unready_on_error(f"Unhandled error: {e}")
226240

227241
finally:
228242
# Stop the geo lookup service if it was started
@@ -278,11 +292,18 @@ async def _resync(self, last_update: bool) -> dict[str, str]:
278292
old_ips_to_cleanup: list[str] = []
279293
has_missing_country = False
280294

281-
stored_neurons = await self.database.get_neurons()
282-
btul.logging.debug(
283-
f"💾 Neurons loaded from Redis: {len(stored_neurons)}",
284-
prefix=self.settings.logging_name,
285-
)
295+
try:
296+
stored_neurons = await self.database.get_neurons()
297+
btul.logging.debug(
298+
f"💾 Neurons loaded from Redis: {len(stored_neurons)}",
299+
prefix=self.settings.logging_name,
300+
)
301+
except Exception as e:
302+
btul.logging.error(
303+
f"❌ Failed to load neurons from Redis: {e}",
304+
prefix=self.settings.logging_name,
305+
)
306+
raise # Re-raise to trigger resync failure handling
286307

287308
mhotkeys = set()
288309

@@ -416,9 +437,16 @@ async def _resync(self, last_update: bool) -> dict[str, str]:
416437
if stale_ips:
417438
sccc.cleanup_rate_limits_for_ips(stale_ips)
418439

419-
not self.settings.dry_run and await self.database.remove_neurons(
420-
stale_neurons
421-
)
440+
try:
441+
not self.settings.dry_run and await self.database.remove_neurons(
442+
stale_neurons
443+
)
444+
except Exception as e:
445+
btul.logging.error(
446+
f"❌ Failed to remove stale neurons from Redis: {e}",
447+
prefix=self.settings.logging_name,
448+
)
449+
raise # Re-raise to trigger resync failure handling
422450

423451
if neurons_to_delete:
424452
btul.logging.debug(
@@ -431,9 +459,16 @@ async def _resync(self, last_update: bool) -> dict[str, str]:
431459
)
432460

433461
# Remove the neurons
434-
not self.settings.dry_run and await self.database.remove_neurons(
435-
neurons_to_delete
436-
)
462+
try:
463+
not self.settings.dry_run and await self.database.remove_neurons(
464+
neurons_to_delete
465+
)
466+
except Exception as e:
467+
btul.logging.error(
468+
f"❌ Failed to remove deleted neurons from Redis: {e}",
469+
prefix=self.settings.logging_name,
470+
)
471+
raise # Re-raise to trigger resync failure handling
437472

438473
# Clean up rate limit data for deleted neuron IPs
439474
deleted_ips = [
@@ -451,9 +486,16 @@ async def _resync(self, last_update: bool) -> dict[str, str]:
451486
f"🧠 Neurons updated: {[n.hotkey for n in updated_neurons]}",
452487
prefix=self.settings.logging_name,
453488
)
454-
not self.settings.dry_run and await self.database.update_neurons(
455-
updated_neurons
456-
)
489+
try:
490+
not self.settings.dry_run and await self.database.update_neurons(
491+
updated_neurons
492+
)
493+
except Exception as e:
494+
btul.logging.error(
495+
f"❌ Failed to update neurons in Redis: {e}",
496+
prefix=self.settings.logging_name,
497+
)
498+
raise # Re-raise to trigger resync failure handling
457499

458500
# Clean up rate limit data for old IPs when neurons changed IP
459501
if old_ips_to_cleanup:
@@ -465,11 +507,18 @@ async def _resync(self, last_update: bool) -> dict[str, str]:
465507

466508
if last_update is None or updated_neurons or neurons_to_delete:
467509
block = await self.subtensor.get_current_block()
468-
not self.settings.dry_run and await self.database.set_last_updated(block)
469-
btul.logging.debug(
470-
f"📅 Last updated block recorded: #{block}",
471-
prefix=self.settings.logging_name,
472-
)
510+
try:
511+
not self.settings.dry_run and await self.database.set_last_updated(block)
512+
btul.logging.debug(
513+
f"📅 Last updated block recorded: #{block}",
514+
prefix=self.settings.logging_name,
515+
)
516+
except Exception as e:
517+
btul.logging.error(
518+
f"❌ Failed to set last updated block in Redis: {e}",
519+
prefix=self.settings.logging_name,
520+
)
521+
raise # Re-raise to trigger resync failure handling
473522

474523
else:
475524
btul.logging.info(
@@ -519,6 +568,7 @@ async def _notify_with_retry(self, ready) -> bool:
519568
"""
520569
Notify with retry logic using adaptive delays based on actual Redis response times.
521570
Returns True if successful, False if failed after all retries.
571+
Marks metagraph as unready if Redis operations fail.
522572
"""
523573
# If already ready, skip all checks
524574
if ready:
@@ -528,36 +578,50 @@ async def _notify_with_retry(self, ready) -> bool:
528578
base_delay = 0.01 # Start with 10ms base
529579

530580
for attempt in range(max_retries):
531-
# Measure how long the consistency check takes
532-
start_time = asyncio.get_event_loop().time()
533-
success = await self._notify_if_needed(ready)
534-
check_duration = asyncio.get_event_loop().time() - start_time
581+
try:
582+
# Measure how long the consistency check takes
583+
start_time = asyncio.get_event_loop().time()
584+
success = await self._notify_if_needed(ready)
585+
check_duration = asyncio.get_event_loop().time() - start_time
535586

536-
if success:
537-
if attempt > 0: # Only log if we actually retried
538-
btul.logging.info(
539-
f"✅ Data consistency verified on retry {attempt + 1} (took {check_duration*1000:.1f}ms)",
587+
if success:
588+
if attempt > 0: # Only log if we actually retried
589+
btul.logging.info(
590+
f"✅ Data consistency verified on retry {attempt + 1} (took {check_duration*1000:.1f}ms)",
591+
prefix=self.settings.logging_name,
592+
)
593+
return True
594+
595+
if attempt < max_retries - 1: # Don't delay on last attempt
596+
# Adaptive delay: wait 2-5x the time it took for the check
597+
# This accounts for Redis load, network latency, etc.
598+
adaptive_delay = max(base_delay, check_duration * (2 + attempt * 1.5))
599+
600+
btul.logging.debug(
601+
f"🔄 Retry {attempt + 1}/{max_retries}: check took {check_duration*1000:.1f}ms, "
602+
f"waiting {adaptive_delay*1000:.1f}ms...",
540603
prefix=self.settings.logging_name,
541604
)
542-
return True
543-
544-
if attempt < max_retries - 1: # Don't delay on last attempt
545-
# Adaptive delay: wait 2-5x the time it took for the check
546-
# This accounts for Redis load, network latency, etc.
547-
adaptive_delay = max(base_delay, check_duration * (2 + attempt * 1.5))
605+
await asyncio.sleep(adaptive_delay)
606+
else:
607+
btul.logging.error(
608+
f"❌ Failed to verify data consistency after {max_retries} attempts "
609+
f"(last check took {check_duration*1000:.1f}ms)",
610+
prefix=self.settings.logging_name,
611+
)
612+
# Mark as unready due to consistency failure
613+
await self._mark_unready_on_error("Data consistency verification failed")
548614

549-
btul.logging.debug(
550-
f"🔄 Retry {attempt + 1}/{max_retries}: check took {check_duration*1000:.1f}ms, "
551-
f"waiting {adaptive_delay*1000:.1f}ms...",
552-
prefix=self.settings.logging_name,
553-
)
554-
await asyncio.sleep(adaptive_delay)
555-
else:
615+
except Exception as e:
556616
btul.logging.error(
557-
f"❌ Failed to verify data consistency after {max_retries} attempts "
558-
f"(last check took {check_duration*1000:.1f}ms)",
617+
f"❌ Redis operation failed during notify retry (attempt {attempt + 1}): {e}",
559618
prefix=self.settings.logging_name,
560619
)
620+
# Mark as unready due to Redis failure
621+
await self._mark_unready_on_error(f"Redis operation failed: {e}")
622+
623+
if attempt < max_retries - 1:
624+
await asyncio.sleep(base_delay * (attempt + 1))
561625

562626
return False
563627

@@ -780,3 +844,23 @@ def _get_country_for_ip(self, ip: str) -> str:
780844

781845
# Fall back to API lookup
782846
return sccc.get_country(ip)
847+
848+
async def _mark_unready_on_error(self, error_reason: str):
849+
"""
850+
Mark metagraph as unready when errors occur and notify listeners.
851+
This ensures downstream services know the metagraph is in a problematic state.
852+
"""
853+
try:
854+
if not self.settings.dry_run:
855+
await self.database.mark_as_unready()
856+
await self.database.notify_state()
857+
btul.logging.warning(
858+
f"🚫 Metagraph marked as unready due to: {error_reason}",
859+
prefix=self.settings.logging_name,
860+
)
861+
except Exception as notify_error:
862+
# Even if we can't mark as unready (e.g., Redis down), log the attempt
863+
btul.logging.error(
864+
f"❌ Failed to mark metagraph as unready (reason: {error_reason}): {notify_error}",
865+
prefix=self.settings.logging_name,
866+
)

0 commit comments

Comments
 (0)