o Lj]@sddlZddlZddlZddlmZmZmZmZddlm Z m Z m Z ddl m Z mZmZmZmZmZmZmZmZddlmZerLddlmZddlmZeeZededfZd e d d d dfd dZ!de"de e Bd dfddZ#GdddZ$GdddZ%GdddZ&dS)N) TYPE_CHECKINGAny AwaitableCallable)record_connection_handoff!record_connection_relaxed_timeoutrecord_maint_notification_count) MaintenanceNotificationMaintenanceStateMaintNotificationsConfigNodeMovingNotificationOSSNodeMigratedNotificationOSSNodeMigratingNotification"_get_maintenance_notification_name"_get_maintenance_notification_type&_should_skip_connection_timeout_update) get_pool_name) RedisCluster))AsyncMaintNotificationsAbstractConnection.messagetaskzasyncio.Task[Any]returncCs>z|}Wn tjyYdSw|rtj||ddSdS)aTask done-callback that surfaces a failed task's exception in the logs. Without it, an unhandled exception in a fire-and-forget task is only reported by asyncio as a noisy "Task exception was never retrieved" warning. Retrieving the exception here suppresses that warning and logs a meaningful error instead. Cancellation is expected and left unlogged. Bind ``message`` with ``functools.partial`` before passing to ``add_done_callback`` (which supplies ``task``). N)exc_info) exceptionasyncioCancelledErrorloggererror)rrexcr_/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/redis/asyncio/maint_notifications.py_log_task_exception!s r! connection notificationc CsttjsdSd}zt|dd}|r|dnd}|r&t|dkr&|dnd}Wn ttt fy5Ynwd}zt|dd}t |rF|}Wn ttfyRYnwt d|d|d|d|dS) N_writersocknameget_resolved_ipz#Handling maintenance notification: , with connection: , connected to ip z, local socket port: ) r isEnabledForloggingDEBUGgetattrget_extra_infolenAttributeErrorOSError TypeErrorcallabledebug)r"r#socket_addresswriter socket_name resolved_ipr'rrr add_debug_log_for_notification4s:   r9c@seZdZdededdfddZ    d#d d Zd$d d Zd%d dZde ddfddZ de ddfddZ d&de dBddfddZde ddfddZdedededdfddZdedededdfdd Zd%d!d"ZdS)'"AsyncMaintNotificationsPoolHandlerpoolconfigrNcCs0||_||_t|_t|_t|_d|_dSN) r;r<set_processed_notifications_scheduled_tasksrLock_lockr")selfr;r<rrr __init__Us   z+AsyncMaintNotificationsPoolHandler.__init__r"rcCs ||_dSr=)r")rCr"rrr set_connectionas z1AsyncMaintNotificationsPoolHandler.set_connectioncCs0t|j|j}|j|_|j|_|j|_d|_|Sr=)r:r;r<r?r@rBr")rCcopyrrr get_handler_for_connectionfs z=AsyncMaintNotificationsPoolHandler.get_handler_for_connectionc f|j4IdHt|jD] }|r|j|qWdIdHdS1IdHs,wYdSr=rBtupler? is_expiredremoverCr#rrr remove_expired_notificationsp .z?AsyncMaintNotificationsPoolHandler.remove_expired_notificationsr#csB|IdHt|tr||IdHdStd|dSNUnhandled notification type: )rN isinstancer handle_node_moving_notificationrrrMrrr handle_notificationvs  z6AsyncMaintNotificationsPoolHandler.handle_notificationc sN|jjs |js dS|j4IdH||jvr% WdIdHdSttjrBt d|d|j d|j r=|j nd|j rJ|j nd}|j j||j||jjo[|jdudIdH|jjru|jduru||jd|j|||j|j|tt|j dIdH|j|WdIdHdS1IdHswYdS)Nz#Handling node MOVING notification: r(r))r#r<moving_address_srcrun_proactive_reconnect pool_name)r<proactive_reconnectis_relaxed_timeouts_enabledrBr?rr*r+r,r4r"r' getpeernamer;apply_moving_notification new_node_host _schedulettlrVhandle_node_moved_notificationrradd)rCr#rUrrr rS~sV      .zBAsyncMaintNotificationsPoolHandler.handle_node_moving_notificationrUc sV|j4IdH|jj|dIdHWdIdHdS1IdHs$wYdS)z Run proactive reconnect for the pool. Active connections are marked for reconnect after they complete the current command. Inactive connections are disconnected and will be connected on next use. N)rU)rBr;rV)rCrUrrr rVs .z:AsyncMaintNotificationsPoolHandler.run_proactive_reconnectc st|}|j4IdH>ttjr*td|d|jd|jr%|jnd|j }|j j }|j j |||dIdHWdIdHdS1IdHsPwYdS)zN Handle the cleanup after a node moving notification expires. Nz5Reverting temporary changes related to notification: r(r))notification_hashreset_relaxed_timeoutreset_host_address)hashrBrr*r+r,r4r"r'r<r[rZr;cleanup_moving_notification)rCr#rcrdrerrr ras&  .zAAsyncMaintNotificationsPoolHandler.handle_node_moved_notificationdelaycallbackargscGsZt|}t|j||g|R}|j|||jj|t t ddS)Nz1Error handling scheduled maintenance notification) rget_running_looptime create_task _run_afterr@rbadd_done_callbackdiscard functoolspartialr!)rCrhrirjdeadlinerrrr r_s z,AsyncMaintNotificationsPoolHandler._schedulerscs<|t}|dkrt|IdH||IdHdS)Nr)rrkrlsleep)rCrsrirj remainingrrr rns z-AsyncMaintNotificationsPoolHandler._run_aftercsB|jsdSt|j}|D]}|q tj|ddiIdHdS)Nreturn_exceptionsT)r@rJcancelrgather)rCtasksrrrr cancel_scheduled_taskss  z9AsyncMaintNotificationsPoolHandler.cancel_scheduled_tasks)r"rrN)rr:rNr=)__name__ __module__ __qualname__rr rDrErGrNr rTr rSstrrVrafloat_ScheduledCallbackr_rnrzrrrr r:Tsf     ?     r:c@sjeZdZdddeddfddZdefdd Zd eddfd d Zd e d eddfddZ de ddfddZ dS)(AsyncMaintNotificationsConnectionHandlerr"rr<rNcCs||_||_dSr=)r"r<)rCr"r<rrr rDs z1AsyncMaintNotificationsConnectionHandler.__init__cCs2t|jdd}|rt|ddrt|jSt|jS)z Get the pool name from the connection's pool handler. Falls back to connection representation if pool is not available. !_maint_notifications_pool_handlerNr;)r-r"rr;repr)rC pool_handlerrrr _get_pool_names   z7AsyncMaintNotificationsConnectionHandler._get_pool_namer#cst|}t|}t|jj|jj|jj|jj|dIdH|dur+td|dS|r9|t j |IdHdS|j |dIdHdS)N)server_address server_portnetwork_peer_addressnetwork_peer_portmaint_notificationrQ)r#) rrrr"hostportrr%handle_maintenance_start_notificationr MAINTENANCE)handle_maintenance_completed_notification)rCr#notification_typerrrr rT$s( zs&   zNAsyncMaintNotificationsConnectionHandler.handle_maintenance_start_notificationkwargscst|jj|jr dSd}|dr|d}t|j|r|nd|jjdd|jdtj |j_|j |rKt |}t | |ddIdHdSdS)Nr#MAINTENANCE_COMPLETEDT)rdFr)rr"rr<getr9reset_tmp_settingsrr NONEreset_received_notificationsrrr)rCrr#rrrr r\s.     zRAsyncMaintNotificationsConnectionHandler.handle_maintenance_completed_notification) r|r}r~r rDrrr rTr rrrrrrr rs$   rc@sfeZdZdZdddeddfddZdd d Zd eddfd d Zd eddfddZ d e ddfddZ dS)!AsyncOSSMaintNotificationsHandleru0 Cluster-wide handler for OSS (open-source) cluster maintenance notifications. Reacts to SMIGRATED (slot migration completed) push notifications and triggers topology re-initialization via nodes_manager.initialize(). Lock discipline: _lock is held across the whole pool mutation, including await initialize(), so the topology refresh and the subsequent connection marking/disconnect happen as one atomic operation — releasing the lock mid-mutation would let other tasks observe partial state. Re-entrancy is not a deadlock risk here even though asyncio.Lock is non-reentrant: initialize() may dispatch commands whose responses carry new push notifications, but handle_notification schedules the actual handling as a separate background task rather than calling into it inline, so the re-entrant arrival never tries to re-acquire the lock on this call stack. The cheap _in_progress/_processed dedup that gates that scheduling runs without the lock — those sets are only mutated from the single event loop. cluster_clientrr<rNcCs2||_||_t|_t|_t|_t|_dSr=) rr<r>r? _in_progressrrArB_background_tasks)rCrr<rrr rDs   z*AsyncOSSMaintNotificationsHandler.__init__c rHr=rI)rCnrrr rNrOz>AsyncOSSMaintNotificationsHandler.remove_expired_notificationsr#cs||jvs ||jvr dS|j|z#t||}|j|||jj |t t dWdSt yD|j |w)Nz7Error handling maintenance notification background task)rr?rbrrkrm_do_handle_notificationrrorprqrrr! Exception)rCr#rrrr rTs*       z5AsyncOSSMaintNotificationsHandler.handle_notificationc sbz(t|tr||IdHntd|W|j|dSW|j|dS|j|wrP)rRr -handle_oss_maintenance_completed_notificationrrrrprMrrr rs z9AsyncOSSMaintNotificationsHandler._do_handle_notificationc s`|IdH|j4IdH||jvr" WdIdHdSttjr0td|g}t}|j D]=\}}| dd\}}|j j j|t|d}|durZ|||D]} | D]} | dd\} } || t| fqbq\q:|j j j|dIdHt|} | |j j j} | D]}}d}||vrt|j}|jD]}||vrt|d|qt|jd|jdIdHd }nttjrtd |jd ||j j jvrt !|"}|j#||$|j#j%|$t&'t(d |st|jd|jdIdHq|j|WdIdHdS1IdHs)wYdS) Nz!Handling SMIGRATED notification: :r&)rr)additional_startup_nodes_infoFzSMIGRATED - mark for reconnectrXTzSMIGRATED: Node z9 not affected by maintenance, skipping mark for reconnectzCError disconnecting free connections after maintenance notification))rNrBr?rr*r+r,r4r>nodes_to_slots_mappingitemsrsplitr nodes_managerget_nodeintrbkeysappend initializeunion nodes_cachevalues_free _connectionsr9mark_for_reconnectrrrnamerrkrmdisconnect_free_connectionsrrorprqrrr!)rCr#raffected_nodes src_address dest_mappingssrc_hostsrc_portsrc_node dest_mapping dest_address dest_host dest_port all_nodes current_nodehandoff_recordedfree_setconnrrrr rs                0zOAsyncOSSMaintNotificationsHandler.handle_oss_maintenance_completed_notificationr{) r|r}r~__doc__r rDrNr rTrr rrrrr r{s* . r)'rrqr+typingrrrr$redis.asyncio.observability.recorderrrrredis.maint_notificationsr r r r r rrrrredis.observability.attributesrredis.asyncio.clusterrredis.asyncio.connectionr getLoggerr|rrrr!objectr9r:rrrrrr s0,      ;m