ڊjadZddlZddlZddlZddlmZddlmZdZdZ dZ dZ dZ d Z d Zd Zd Zd ZdZdZdZdZdZdZdZdZe ee ee ee eeeeeeeeeiZej7Dcic]\}}|| c}}ZdZdZdZd Z d Z!dZ"dZ#dZ$dZ%dZ&dZ'dZ(dZ)ejTe)Z+dZ,GddZ-e-Z. d.de/d e/d!e0d"e1d#e1f d$Z2d#e1fd%Z3d#e1fd&Z4d#e1fd'Z5d#e1fd(Z6 d/d)e7d*e/d#e1fd+Z8d0d)e7d,e7d#e1fd-Z9ycc}}w)1a6 Dirty Worker Binary Protocol Binary message framing over Unix sockets, inspired by OpenBSD msgctl/msgsnd. Replaces JSON protocol for efficient binary data transfer. Header Format (16 bytes): +--------+--------+--------+--------+--------+--------+--------+--------+ | Magic (2B) | Ver(1) | MType | Payload Length (4B) | +--------+--------+--------+--------+--------+--------+--------+--------+ | Request ID (8 bytes) | +--------+--------+--------+--------+--------+--------+--------+--------+ - Magic: 0x47 0x44 ("GD" for Gunicorn Dirty) - Version: 0x01 - MType: Message type (REQUEST, RESPONSE, ERROR, CHUNK, END) - Length: Payload size (big-endian uint32, max 64MB) - Request ID: uint64 (replaces UUID string) Payload is TLV-encoded (see tlv.py). N)DirtyProtocolError) TLVEncodersGDrequestresponseerrorchunkendstashstatusmanage z>2sBBIQiceZdZdZeZeZeZeZ e Z e Z eZeZeZeZededededefdZededefdZe d%ded ed ed ed edef dZededefdZededefdZ ededefdZ!ededefdZ"ededefdZ#ed&dedededefdZ$e d'dedededefdZ%ededefdZ&ede'jPdefdZ)ede'jTdedd fdZ+ede,jXd edefd!Z-ede,jXdefd"Z.ede,jXdedd fd#Z/ededefd$Z0y )(BinaryProtocolz-Binary message protocol for dirty worker IPC.msg_type request_idpayload_lengthreturncNtjttt|||S)a, Encode the 16-byte message header. Args: msg_type: Message type (MSG_TYPE_REQUEST, etc.) request_id: Unique request identifier (uint64) payload_length: Length of the TLV-encoded payload Returns: bytes: 16-byte header )structpack HEADER_FORMATMAGICVERSION)rrrs \/var/www/io.vulcan-creative.com/venv/lib/python3.12/site-packages/gunicorn/dirty/protocol.py encode_headerzBinaryProtocol.encode_headerrs"{{=%():7 7datact|tkr tdt|dt|tjt |dt\}}}}}|t k7rtd|dt |dd|tk7rtd|dt|dd|tvrtd |d |dd|tkDrtd |d td |||fS)z Decode the 16-byte message header. Args: data: 16 bytes of header data Returns: tuple: (msg_type, request_id, payload_length) Raises: DirtyProtocolError: If header is invalid zHeader too short:  bytes, expected raw_dataNzInvalid magic: z , expected zUnsupported protocol version: zUnknown message type: 0x02xzMessage too large: z bytes (max: )) len HEADER_SIZErr!unpackr#r$r%MSG_TYPE_TO_STRMAX_MESSAGE_SIZE)r)magicversionrlengthrs r& decode_headerzBinaryProtocol.decode_headers" t9{ "$$SYK/@ N  8>}} 4 -8 4w&* E>$!%+eY?cr  g $0 WINcr  ? *$*8C.9cr  $ $$%fX];K:LAN V++r(Napp_pathactionargskwargsc|||r t|ng|xsid}tj|}tj t |t |}||zS)al Encode a request message. Args: request_id: Unique request identifier (uint64) app_path: Import path of the dirty app action: Action to call on the app args: Positional arguments kwargs: Keyword arguments Returns: bytes: Complete message (header + payload) )r:r;r<r=)listrencoderr'MSG_TYPE_REQUESTr1)rr:r;r<r= payload_dictpayloadheaders r&encode_requestzBinaryProtocol.encode_requests\"!"&DJBl  ##L1--.> .1'l<r(cd|i}tj|}tjt|t |}||zS)a Encode a success response message. Args: request_id: Request identifier this responds to result: Result value (must be TLV-serializable) Returns: bytes: Complete message (header + payload) result)rr@rr'MSG_TYPE_RESPONSEr1)rrGrBrCrDs r&encode_responsezBinaryProtocol.encode_responsesE!&) ##L1--.?.1'l<r(c6ddlm}t||r|j}n5t|tr|}n"t |j t|id}d|i}tj|}tjt|t|}||zS)a Encode an error response message. Args: request_id: Request identifier this responds to error: DirtyError instance, dict, or Exception Returns: bytes: Complete message (header + payload) r DirtyError error_typemessagedetailsr)errorsrL isinstanceto_dictdicttype__name__strrr@rr'MSG_TYPE_ERRORr1)rrrL error_dictrBrCrDs r& encode_errorzBinaryProtocol.encode_errors ' eZ (J t $J#5k22u:J  , ##L1--nj.1'l<r(cd|i}tj|}tjt|t |}||zS)a Encode a chunk message for streaming responses. Args: request_id: Request identifier this chunk belongs to data: Chunk data (must be TLV-serializable) Returns: bytes: Complete message (header + payload) r))rr@rr'MSG_TYPE_CHUNKr1)rr)rBrCrDs r& encode_chunkzBinaryProtocol.encode_chunksC~ ##L1--nj.1'l<r(c<tjt|d}|S)z Encode an end-of-stream message. Args: request_id: Request identifier this ends Returns: bytes: Complete message (header + empty payload) r)rr' MSG_TYPE_ENDrrDs r& encode_endzBinaryProtocol.encode_ends --lJJ r(c<tjt|d}|S)z Encode a status query message. Args: request_id: Request identifier Returns: bytes: Complete message (header + empty payload) r)rr'MSG_TYPE_STATUSr`s r& encode_statuszBinaryProtocol.encode_status s --oz1M r(opcountc||d}tj|}tjt|t |}||zS)a4 Encode a worker management message. Args: request_id: Request identifier op: Management operation (MANAGE_OP_ADD or MANAGE_OP_REMOVE) count: Number of workers to add/remove Returns: bytes: Complete message (header + payload) )rerf)rr@rr'MSG_TYPE_MANAGEr1)rrerfrBrCrDs r& encode_managezBinaryProtocol.encode_manage/sI ##L1--oz.1'l<r(tablec||d}|||d<|||d<|||d<tj|}tjt|t |}||zS)a Encode a stash operation message. Args: request_id: Unique request identifier (uint64) op: Stash operation code (STASH_OP_*) table: Table name key: Optional key for put/get/delete operations value: Optional value for put operation pattern: Optional pattern for keys operation Returns: bytes: Complete message (header + payload) )rerjkeyvaluepattern)rr@rr'MSG_TYPE_STASHr1) rrerjrlrmrnrBrCrDs r& encode_stashzBinaryProtocol.encode_stashEsx$  ?"%L   $)L !  &-L ###L1--nj.1'l<r(c|tj|\}}}t|t|zkr&t dt|zdt||dd|dk(ri}n&|tt|z} t j |}t|}|||fS#t$rt$r}t d||ddd}~wwxYw)a Decode a complete message (header + payload). Args: data: Complete message bytes Returns: tuple: (msg_type_str, request_id, payload_dict) msg_type_str is the string name (e.g., "request") payload_dict is the decoded TLV payload as a dict Raises: DirtyProtocolError: If message is malformed zIncomplete message: expected z bytes, got N2r,rFailed to decode TLV payload: ) rr9r1r2rr decode_full Exceptionr4)r)rrr8rB payload_datae msg_type_strs r&decode_messagezBinaryProtocol.decode_messagefs (6'C'CD'I$*f t9{V+ +$/ f0D/EF4yk#cr  Q;L K&,@AL )55lC 'x0 Z55&  (4QC8)#2. s.BB;#B66B;readercK |jtd{}tj|\}}}|dkDr1 |j|d{} tj|}ni}t|}||d } | j|| S7v#tj$rR}t |j dk(rt dt |j dt|j d}~wwxYw7#tj$r5}t dt |j d||j d}~wwxYw#t $rt$r}t d||ddd}~wwxYww) ap Read a complete binary message from async stream. Args: reader: asyncio StreamReader Returns: dict: Message dict with 'type', 'id', and payload fields Raises: DirtyProtocolError: If read fails or message is malformed asyncio.IncompleteReadError: If connection closed mid-read NrzIncomplete header: got r+r,zIncomplete payload: got rsrrrUid) readexactlyr2asyncioIncompleteReadErrorr1partialrrr9rrtrur4update) rzrDrwrrr8rvrBrxrGs r&read_message_asyncz!BinaryProtocol.read_message_asyncs !--k::F(6'C'CF'K$*f A: %+%7%7%??  )55lC L'x0 &j9 l# S;** 199~"$)#aii.)9:'=*    @.. (.s199~.>? &x)YY &  (4QC8)#2. sE7BBBE7C?C=C?E /$E7BC:(A C55C::E7=C??E0EEE7 E4E//E44E7writerrOcKtj|}|j||jd{y7w)a[ Write a message to async stream. Accepts dict format for backwards compatibility. Args: writer: asyncio StreamWriter message: Message dict with 'type', 'id', and payload fields Raises: DirtyProtocolError: If encoding fails ConnectionError: If write fails N)r_encode_from_dictwritedrain)rrOr)s r&write_message_asyncz"BinaryProtocol.write_message_asyncs3 //8 Tllns:AAAsockncd}t||krh|j|t|z }|s5t|dk(r tdtdt|d||||z }t||krh|S)a Receive exactly n bytes from a socket. Args: sock: Socket to read from n: Number of bytes to read Returns: bytes: Received data Raises: DirtyProtocolError: If read fails or connection closed r(rzConnection closedzConnection closed after r+r,)r1recvr)rrr)rs r& _recv_exactlyzBinaryProtocol._recv_exactlys$i!mIIa#d)m,Et9>,-@AA(.s4yk9J1#N! EMD$i!m r(cltj|t}tj|\}}}|dkDr-tj||} t j |}ni}t|}||d} | j|| S#t $rt$r}t d||ddd}~wwxYw)a  Read a complete message from socket (sync). Args: sock: Socket to read from Returns: dict: Message dict with 'type', 'id', and payload fields Raises: DirtyProtocolError: If read fails or message is malformed rrsNrrr,r|) rrr2r9rrtrrur4r) rrDrrr8rvrBrwrxrGs r& read_messagezBinaryProtocol.read_messages --dK@'5'C'CF'K$*f A:)77fEL )55lC L'x0 &j9 l# &  (4QC8)#2. sB B3B..B3cPtj|}|j|y)a Write a message to socket (sync). Args: sock: Socket to write to message: Message dict with 'type', 'id', and payload fields Raises: DirtyProtocolError: If encoding fails OSError: If write fails N)rrsendall)rrOr)s r& write_messagezBinaryProtocol.write_message/s //8 Tr(c |jd}|jdd}t|trt|dz}tj|}|t d||t k(rWtj||jdd|jdd|jd |jd S|tk(r%tj||jd S|tk(r&tj||jd iS|tk(r%tj||jd S|tk(rtj!|S|t"k(rftj%||jd|jdd|jd|jd|jdS|t&k(rtj)|S|t*k(r6tj-||jd|jddSt d|)a Encode a message dict to binary format. Supports the old dict-based API for backwards compatibility. Args: message: Message dict with 'type', 'id', and payload fields Returns: bytes: Complete encoded message rUr}rlzUnknown message type: r:r;r<r=rGrr)rerjrlrmrnrfrzUnhandled message type: )getrRrWhashMSG_TYPE_FROM_STRrrArrErHrIrXrZr\r]r_rarorprcrdrhri)rOrxrrs r&rz BinaryProtocol._encode_from_dict?s {{6* [[q)  j# &j),>>J$((6  $'=l^%LM M ' '!00 J+ Hb) F# H%  * *!11 H%  '!.. GR(  '!.. F#  %!,,Z8 8  '!.. D! GR( E" G$ I&   (!// ; ;  (!// D! GQ'  %'?z%JK Kr(NNrNNN)1rV __module__ __qualname____doc__r2r5MSG_TYPE_REQUEST_STRrAMSG_TYPE_RESPONSE_STRrHMSG_TYPE_ERROR_STRrXMSG_TYPE_CHUNK_STRr\MSG_TYPE_END_STRr_MSG_TYPE_STASH_STRroMSG_TYPE_STATUS_STRrcMSG_TYPE_MANAGE_STRrh staticmethodintbytesr'tupler9rWrTrErIrZr]rardrirpryr StreamReaderr StreamWriterrsocketrrrrr(r&rrbs7K'+-'N'N#L'N)O)O 7 7 7c 7e 7 7.,E.,e.,.,`:> 3 # s " 37 CH  4 C E  "    <  u  " s u   # %   # 3 s 5  *37  # c *>+/48.FMMce6%6==%T%%N FMM D T  CL4CLECLCLr(rr:r;r<r=rcRtj||||r t|ng|xsidS)aU Build a request message dict. Args: request_id: Unique request identifier (int or str) app_path: Import path of the dirty app (e.g., 'myapp.ml:MLApp') action: Action to call on the app args: Positional arguments kwargs: Keyword arguments Returns: dict: Request message dict )rUr}r:r;r<r=) DirtyProtocolrAr?)rr:r;r<r=s r& make_requestrs2 .."T ,B  r(c*tj||dS)z Build a success response message dict. Args: request_id: Request identifier this responds to result: Result value Returns: dict: Response message dict )rUr}rG)rrH)rrGs r& make_responsers// r(cddlm}t||r|j}n5t|tr|}n"t |j t|id}tj||dS)z Build an error response message dict. Args: request_id: Request identifier this responds to error: DirtyError instance or dict with error info Returns: dict: Error response message dict rrKrM)rUr}r) rQrLrRrSrTrUrVrWrrX)rrrLrYs r&make_error_responsersf#%$]]_ E4  u+..5z ,, r(c*tj||dS)z Build a chunk message dict for streaming responses. Args: request_id: Request identifier this chunk belongs to data: Chunk data Returns: dict: Chunk message dict )rUr}r))rr\)rr)s r&make_chunk_messagers,, r(c(tj|dS)z Build an end-of-stream message dict. Args: request_id: Request identifier this ends Returns: dict: End message dict r|)rr_)rs r&make_end_messagers** r(rerjcZtj|||d}|||d<|||d<|||d<|S)a Build a stash operation message dict. Args: request_id: Unique request identifier (int or str) op: Stash operation code (STASH_OP_*) table: Table name key: Optional key for put/get/delete operations value: Optional value for put operation pattern: Optional pattern for keys operation Returns: dict: Stash message dict )rUr}rerjrlrmrn)rro)rrerjrlrmrnmsgs r&make_stash_messagersP",,  C  E  G  I Jr(rfc,tj|||dS)a Build a worker management message dict. Args: request_id: Unique request identifier (int or str) op: Management operation (MANAGE_OP_ADD or MANAGE_OP_REMOVE) count: Number of workers to add/remove Returns: dict: Manage message dict )rUr}rerf)rrh)rrerfs r&make_manage_messagers --  r(rrr):rrrr!rQrtlvrr$r%rArHrXr\r_rorcrhrrrrrrrrr4itemsr STASH_OP_PUT STASH_OP_GETSTASH_OP_DELETE STASH_OP_KEYSSTASH_OP_CLEAR STASH_OP_INFOSTASH_OP_ENSURESTASH_OP_DELETE_TABLESTASH_OP_TABLESSTASH_OP_EXISTS MANAGE_OP_ADDMANAGE_OP_REMOVEr#calcsizer2r5rrrWrrTrrrrrrrr)kvs00r&rs , &   !"*,&&"&(( '6&;&;&=>daQT>      foom, $aLaLR 59sC-1=A2$d<D$ D "6:s3>B>Cc?s$ C/