From e0f6ed78abb49380a21b232ee09bad1e23b40dbc Mon Sep 17 00:00:00 2001 From: Tom J Nowell Date: Tue, 15 Sep 2026 23:28:50 +0100 Subject: [PATCH] Put a command's #id on its answer and nothing else The id stayed on the Client after the handler returned, so chat and broadcasts carried it until the next command, while replies from DB callbacks carried whichever id came later. _handle now restores the id when it returns, and callbacks that answer the client put back the id captured when the work was deferred. Offline messages delivered after LOGIN are left without one, because they are other players' messages rather than the answer. Fixes #60. --- Client.py | 14 ++++ DataHandler.py | 6 +- PROTOCOL.md | 10 ++- protocol/Protocol.py | 132 ++++++++++++++++--------------- tests/integration/msgidtest.py | 138 +++++++++++++++++++++++++++++++++ 5 files changed, 231 insertions(+), 69 deletions(-) create mode 100644 tests/integration/msgidtest.py diff --git a/Client.py b/Client.py index 927502e6..dcb60863 100644 --- a/Client.py +++ b/Client.py @@ -153,6 +153,20 @@ def set_msg_id(self, msg): self.msg_id = '#%s ' % test return (' '.join(msg.split(' ')[1:])) + def with_msg_id(self, callback): + # a reply sent after the handler returns, from a DB callback or the login queue, still + # answers the command that started it. Capture that command's id now and put it back + # on the client only while the callback runs. + msg_id = self.msg_id + def run(*args, **kwargs): + previous = self.msg_id + self.msg_id = msg_id + try: + return callback(*args, **kwargs) + finally: + self.msg_id = previous + return run + def setFlagByIP(self, ip, force=True): cc = ip2country.lookup(ip) if force or cc != '??': diff --git a/DataHandler.py b/DataHandler.py index 5c462a2d..dcd1e11c 100644 --- a/DataHandler.py +++ b/DataHandler.py @@ -210,7 +210,7 @@ def __init__(self): # simultaneously backed up on their write buffers (the 2.3 producer signal), new # logins are queued FIFO and drained by drain_login_queue() (a 1s LoopingCall) once # the backpressure clears. Under normal load the queue stays empty. - self.login_queue = collections.deque() # (client, login_args) awaiting login under backpressure + self.login_queue = collections.deque() # (client, login_now carrying the LOGIN's msg_id, login_args) awaiting login under backpressure self.login_backpressure_limit = 50 # paused-producer count above which login admission pauses # rate limits @@ -807,11 +807,11 @@ def drain_login_queue(self): # shares login_queue with in_LOGIN without locking. Each login gets its own session # commit/rollback/close, mirroring the per-request guards in dataReceived. while self.login_queue and not self.login_backpressured(): - client, args = self.login_queue.popleft() + client, login_now, args = self.login_queue.popleft() if client.session_id not in self.clients: continue # client disconnected while queued try: - self.protocol.login_now(client, *args) + login_now(client, *args) self.session_manager.commit_guard() except: logging.error(traceback.format_exc()) diff --git a/PROTOCOL.md b/PROTOCOL.md index 6dfa6f68..4b3139d1 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -74,9 +74,13 @@ cannot drift. **[GAP]** the `@emits` convention does not exist yet — see argument may legitimately contain spaces (e.g. chat text, topics). This is the "sentence argument" behaviour. (`Protocol.py` `get_function_args`.) - **Message IDs.** A client may prefix a command with `# ` to correlate a request - with the server's reply; the server echoes the id back on responses generated in that - command's handling. (`Client.py`.) **[GAP]** document exact echo semantics and which - responses carry the id vs which do not. + with the server's reply. The server puts the same `# ` on every line it sends that + client as the answer to that command, including refusals (`SERVERMSG`, `FAILED`) and + replies that arrive later because the command waited on the database or the login queue. + Lines that are not an answer to the command carry no id, even when they arrive before + the answer: chat from other players, broadcasts, status changes, and offline messages + delivered after `LOGIN`. Lines sent to other clients as a result of the command carry no + id either. (`Client.py` `set_msg_id`, `with_msg_id`.) - **Tab-separated payloads.** Some structured fields (notably battle `script_tags`) use tab-separated `key=value` pairs. **[GAP]** enumerate every command that uses tab separation and the exact field grammar. diff --git a/protocol/Protocol.py b/protocol/Protocol.py index 99919a87..8e9a06a5 100755 --- a/protocol/Protocol.py +++ b/protocol/Protocol.py @@ -395,11 +395,17 @@ def get_function_args(self, client, command, function, numspaces, args): def _handle(self, client, msg): assert(type(msg) == str) - # client.Send() prepends client.msg_id if the current thread - # is the same thread as the client's handler. - # this works because handling is done in order for each ClientHandler thread - # so we can be sure client.Send() was performed in the client's own handling code. + # client.Send() prepends client.msg_id, so the id is set only while this command is + # handled. Restore rather than clear: ChanServ runs a mod command typed into #moderator + # through here while the same client's SAY is still being handled. + previous_msg_id = client.msg_id msg = client.set_msg_id(msg) + try: + return self._dispatch(client, msg) + finally: + client.msg_id = previous_msg_id + + def _dispatch(self, client, msg): numspaces = msg.count(' ') if (numspaces > 0): @@ -1019,8 +1025,8 @@ def in_REGISTER(self, client, username, password, email = ''): # the INSERT continue in _register_checked, preserving the original denial ordering. email = email.lower() d = self._root.defer_db(self.userdb.check_register_user, username, email, client.ip_address) - d.addCallback(self._register_checked, client, username, password, email) - d.addErrback(self._register_failed, client, username) + d.addCallback(client.with_msg_id(self._register_checked), client, username, password, email) + d.addErrback(client.with_msg_id(self._register_failed), client, username) def _register_checked(self, verdict, client, username, password, email): # reactor-thread callback after the uniqueness/ban check. Applies the same memory-only @@ -1054,8 +1060,8 @@ def _register_checked(self, verdict, client, username, password, email): # hop 2: INSERT off the reactor; the worker catches the unique-constraint race. d = self._root.defer_db(self.userdb.do_register_insert, username, password, client.ip_address, email) - d.addCallback(self._register_inserted, client, username, email) - d.addErrback(self._register_failed, client, username) + d.addCallback(client.with_msg_id(self._register_inserted), client, username, email) + d.addErrback(client.with_msg_id(self._register_failed), client, username) def _register_inserted(self, verdict, client, username, email): # reactor-thread callback after the INSERT. @@ -1152,7 +1158,7 @@ def in_LOGIN(self, client, username, password, cpu='0', local_ip='', sentence_ar keeps the same signature so command dispatch (which reflects on it) is unchanged. ''' if self._root.login_queue or self._root.login_backpressured(): - self._root.login_queue.append((client, (username, password, cpu, local_ip, sentence_args))) + self._root.login_queue.append((client, client.with_msg_id(self.login_now), (username, password, cpu, local_ip, sentence_args))) client.Send('SERVERMSG The server is busy; you are number %d in the login queue, please wait...' % len(self._root.login_queue)) return self.login_now(client, username, password, cpu, local_ip, sentence_args) @@ -1193,8 +1199,8 @@ def login_now(self, client, username, password, cpu='0', local_ip='', sentence_a # _login_checked once the DB worker returns. The remaining memory-only checks and # the denial ordering are applied in the callback, preserving the original flow. d = self._root.defer_db(self.userdb.precheck_login, username, password, client.ip_address) - d.addCallback(self._login_checked, client, username, local_ip, sentence_args) - d.addErrback(self._login_failed, client, username) + d.addCallback(client.with_msg_id(self._login_checked), client, username, local_ip, sentence_args) + d.addErrback(client.with_msg_id(self._login_failed), client, username) def _login_checked(self, result, client, username, local_ip, sentence_args): # reactor-thread callback after precheck_login (result is plain data). Applies the @@ -1239,8 +1245,8 @@ def _login_checked(self, result, client, username, local_ip, sentence_args): # login checks complete; write the login record off the reactor thread. d = self._root.defer_db(self.userdb.do_login, username, client.ip_address, agent, last_sys_id, last_mac_id, local_ip, client.country_code) - d.addCallback(self._login_finish, client, username, agent, local_ip) - d.addErrback(self._login_failed, client, username) + d.addCallback(client.with_msg_id(self._login_finish), client, username, agent, local_ip) + d.addErrback(client.with_msg_id(self._login_failed), client, username) def _login_finish(self, result, client, username, agent, local_ip): # reactor-thread callback after do_login. result is (snapshot, ignored_user_ids): @@ -1466,8 +1472,8 @@ def in_CONFIRMAGREEMENT(self, client, verification_code = ""): # dump. _SendLoginInfo touches the DB on the reactor (get_ignored_user_ids), so the # callback brackets it with the session guards exactly like _login_finish does. d = self._root.defer_db(self.userdb.do_confirm_agreement, client.username) - d.addCallback(self._confirmagreement_done, client) - d.addErrback(self._confirmagreement_failed, client) + d.addCallback(client.with_msg_id(self._confirmagreement_done), client) + d.addErrback(client.with_msg_id(self._confirmagreement_failed), client) def _confirmagreement_done(self, uid, client): if client.session_id not in self._root.clients: @@ -1675,8 +1681,8 @@ def _enqueue_offline_message(self, client, user, msg, ex_msg): # never echo a message that was refused (unknown user, bot target). cmd = 'SAYPRIVATEEX' if ex_msg else 'SAYPRIVATE' d = self._root.defer_db(self.userdb.do_enqueue_offline_message, client.user_id, user, msg, ex_msg) - d.addCallback(self._offline_message_queued, client, user, msg, cmd) - d.addErrback(self._offline_message_failed, client, user, cmd) + d.addCallback(client.with_msg_id(self._offline_message_queued), client, user, msg, cmd) + d.addErrback(client.with_msg_id(self._offline_message_failed), client, user, cmd) def _offline_message_queued(self, result, client, user, msg, cmd): # reactor-thread callback: pure send, no DB -> no commit/rollback/close bracket. @@ -1937,8 +1943,8 @@ def in_IGNORE(self, client, tags): # target's access; the already-ignored/list-full checks and the client.ignored # mutation stay on the reactor thread (live memory) - see the callbacks below. d = self._root.defer_db(self.userdb._user_access_from_username, username) - d.addCallback(self._ignore_resolved, client, username, reason) - d.addErrback(self._ignore_op_failed, client, "IGNORE") + d.addCallback(client.with_msg_id(self._ignore_resolved), client, username, reason) + d.addErrback(client.with_msg_id(self._ignore_op_failed), client, "IGNORE") def _ignore_resolved(self, resolved, client, username, reason): # reactor-thread callback: resolved is (target_id, access) or None. No reactor-side DB @@ -1964,8 +1970,8 @@ def _ignore_resolved(self, resolved, client, username, reason): return # checks passed: do the bare INSERT off the reactor, then sync memory + reply. d = self._root.defer_db(self.userdb.do_ignore_insert, client.user_id, target_id, reason) - d.addCallback(self._ignore_stored, client, username, target_id, reason) - d.addErrback(self._ignore_op_failed, client, "IGNORE") + d.addCallback(client.with_msg_id(self._ignore_stored), client, username, target_id, reason) + d.addErrback(client.with_msg_id(self._ignore_op_failed), client, "IGNORE") def _ignore_stored(self, result, client, username, target_id, reason): # reactor-thread callback: the row is committed; sync in-memory state + reply. @@ -1992,8 +1998,8 @@ def in_UNIGNORE(self, client, tags): # 3.1: resolve off the reactor (reuse the IGNORE resolver; access is unused here). # The not-ignored check reads live memory and the DELETE runs off the reactor. d = self._root.defer_db(self.userdb._user_access_from_username, username) - d.addCallback(self._unignore_resolved, client, username) - d.addErrback(self._ignore_op_failed, client, "UNIGNORE") + d.addCallback(client.with_msg_id(self._unignore_resolved), client, username) + d.addErrback(client.with_msg_id(self._ignore_op_failed), client, "UNIGNORE") def _unignore_resolved(self, resolved, client, username): # reactor-thread callback: only memory + a second deferred, so no session bracket. @@ -2007,8 +2013,8 @@ def _unignore_resolved(self, resolved, client, username): self.out_SERVERMSG(client, "User is not ignored.") return d = self._root.defer_db(self.userdb.do_unignore_delete, client.user_id, target_id) - d.addCallback(self._unignore_removed, client, username, target_id) - d.addErrback(self._ignore_op_failed, client, "UNIGNORE") + d.addCallback(client.with_msg_id(self._unignore_removed), client, username, target_id) + d.addErrback(client.with_msg_id(self._ignore_op_failed), client, "UNIGNORE") def _unignore_removed(self, result, client, username, target_id): # reactor-thread callback: the row(s) are deleted; sync in-memory state + reply. @@ -2029,8 +2035,8 @@ def in_IGNORELIST(self, client): # 3.1: read the ignore list off the reactor thread, then format + send in # _ignorelist_send once the worker returns plain [(ignored_user_id, reason), ...]. d = self._root.defer_db(self.userdb.get_ignore_list, client.user_id) - d.addCallback(self._ignorelist_send, client) - d.addErrback(self._social_list_failed, client, "IGNORELIST") + d.addCallback(client.with_msg_id(self._ignorelist_send), client) + d.addErrback(client.with_msg_id(self._social_list_failed), client, "IGNORELIST") def _ignorelist_send(self, entries, client): # reactor-thread callback. clientFromID may do a reactor-side DB read on an @@ -2078,8 +2084,8 @@ def in_FRIENDREQUEST(self, client, tags): # online ignore-set is kept in sync with the ignores table, so the db check is # correct either way. Notify an online target in the callback. d = self._root.defer_db(self.userdb.do_friend_request, client.user_id, username, msg) - d.addCallback(self._friendrequest_done, client, username, msg) - d.addErrback(self._friend_op_failed, client, "FRIENDREQUEST") + d.addCallback(client.with_msg_id(self._friendrequest_done), client, username, msg) + d.addErrback(client.with_msg_id(self._friend_op_failed), client, "FRIENDREQUEST") def _friendrequest_done(self, verdict, client, username, msg): # reactor-thread callback: no reactor-side DB (clientFromID is memory-only here), @@ -2116,8 +2122,8 @@ def in_ACCEPTFRIENDREQUEST(self, client, tags): # 3.1: verify the request exists + create the friendship + delete the request as ONE # atomic DB unit off the reactor; notify an online requester in the callback. d = self._root.defer_db(self.userdb.do_accept_friend_request, client.user_id, username) - d.addCallback(self._acceptfriend_done, client, username) - d.addErrback(self._friend_op_failed, client, "ACCEPTFRIENDREQUEST") + d.addCallback(client.with_msg_id(self._acceptfriend_done), client, username) + d.addErrback(client.with_msg_id(self._friend_op_failed), client, "ACCEPTFRIENDREQUEST") def _acceptfriend_done(self, verdict, client, username): if client.session_id not in self._root.clients: @@ -2141,8 +2147,8 @@ def in_DECLINEFRIENDREQUEST(self, client, tags): return # 3.1: verify + delete the request as ONE atomic DB unit off the reactor. d = self._root.defer_db(self.userdb.do_decline_friend_request, client.user_id, username) - d.addCallback(self._declinefriend_done, client) - d.addErrback(self._friend_op_failed, client, "DECLINEFRIENDREQUEST") + d.addCallback(client.with_msg_id(self._declinefriend_done), client) + d.addErrback(client.with_msg_id(self._friend_op_failed), client, "DECLINEFRIENDREQUEST") def _declinefriend_done(self, verdict, client): if client.session_id not in self._root.clients: @@ -2160,8 +2166,8 @@ def in_UNFRIEND(self, client, tags): # 3.1: resolve + delete the friendship (both directions) as ONE atomic DB unit off # the reactor; notify an online ex-friend in the callback. d = self._root.defer_db(self.userdb.do_unfriend, client.user_id, username) - d.addCallback(self._unfriend_done, client, username) - d.addErrback(self._friend_op_failed, client, "UNFRIEND") + d.addCallback(client.with_msg_id(self._unfriend_done), client, username) + d.addErrback(client.with_msg_id(self._friend_op_failed), client, "UNFRIEND") def _unfriend_done(self, verdict, client, username): if client.session_id not in self._root.clients: @@ -2188,8 +2194,8 @@ def _friend_op_failed(self, failure, client, op): def in_FRIENDREQUESTLIST(self, client): # 3.1: read off the reactor thread; worker returns plain [(user_id, msg), ...]. d = self._root.defer_db(self.userdb.get_friend_request_list, client.user_id) - d.addCallback(self._friendrequestlist_send, client) - d.addErrback(self._social_list_failed, client, "FRIENDREQUESTLIST") + d.addCallback(client.with_msg_id(self._friendrequestlist_send), client) + d.addErrback(client.with_msg_id(self._social_list_failed), client, "FRIENDREQUESTLIST") def _friendrequestlist_send(self, entries, client): # reactor-thread callback; bracket the session because clientFromID may read the DB. @@ -2215,8 +2221,8 @@ def _friendrequestlist_send(self, entries, client): def in_FRIENDLIST(self, client): # 3.1: read off the reactor thread; worker returns plain [user_id, ...]. d = self._root.defer_db(self.userdb.get_friend_user_ids, client.user_id) - d.addCallback(self._friendlist_send, client) - d.addErrback(self._social_list_failed, client, "FRIENDLIST") + d.addCallback(client.with_msg_id(self._friendlist_send), client) + d.addErrback(client.with_msg_id(self._social_list_failed), client, "FRIENDLIST") def _friendlist_send(self, userIds, client): # reactor-thread callback; bracket the session because clientFromID may read the DB. @@ -2817,8 +2823,8 @@ def in_GETCHANNELMESSAGES(self, client, chan, last_msg_id): # query (joins to User/BridgedUser), so the callback does NO reactor-side DB and # needs no session bracket; it just formats + sends. A pure read, so no new races. d = self._root.defer_db(self.userdb.get_channel_messages, client.user_id, channel.id, last_msg_id, self._channel_history_max_fetch) - d.addCallback(self._channelmessages_send, client, chan) - d.addErrback(self._channelmessages_failed, client) + d.addCallback(client.with_msg_id(self._channelmessages_send), client, chan) + d.addErrback(client.with_msg_id(self._channelmessages_failed), client) def _channelmessages_send(self, result, client, chan): # reactor-thread callback: pure send, no DB -> no commit/rollback/close bracket. @@ -3196,8 +3202,8 @@ def in_FINDIP(self, client, address): # plain (username, last_login) tuples. The online/offline check reads shared memory, so # it stays in the reactor callback. d = self._root.defer_db(self.userdb.do_find_ip, address) - d.addCallback(self._findip_done, client, address) - d.addErrback(self._findip_failed, client, address) + d.addCallback(client.with_msg_id(self._findip_done), client, address) + d.addErrback(client.with_msg_id(self._findip_failed), client, address) def _findip_done(self, results, client, address): if client.session_id not in self._root.clients: @@ -3231,8 +3237,8 @@ def in_GETIP(self, client, username): # 3.1: only the offline branch hits the DB (get_ip returns a plain string); defer it. # The online check above is pure memory and stays on the reactor. d = self._root.defer_db(self.userdb.get_ip, username) - d.addCallback(self._getip_done, client, username) - d.addErrback(self._getip_failed, client, username) + d.addCallback(client.with_msg_id(self._getip_done), client, username) + d.addErrback(client.with_msg_id(self._getip_failed), client, username) def _getip_done(self, ip, client, username): if client.session_id not in self._root.clients: @@ -3274,8 +3280,8 @@ def in_RENAMEACCOUNT(self, client, newname): # the old and new name, replies, and disconnects on success; it does no reactor-side DB, # so it needs no session bracket. d = self._root.defer_db(self.userdb.do_rename_account, client.username, newname) - d.addCallback(self._renameaccount_done, client, newname) - d.addErrback(self._renameaccount_failed, client, newname) + d.addCallback(client.with_msg_id(self._renameaccount_done), client, newname) + d.addErrback(client.with_msg_id(self._renameaccount_failed), client, newname) def _renameaccount_done(self, verdict, client, newname): if client.session_id not in self._root.clients: @@ -3315,8 +3321,8 @@ def in_CHANGEPASSWORD(self, client, cur_password, new_password): # as one atomic DB unit. The callback (on the reactor) invalidates the 1.2 user # cache and replies; it does no reactor-side DB, so it needs no session bracket. d = self._root.defer_db(self.userdb.do_change_password, client.username, cur_password, new_password) - d.addCallback(self._changepassword_done, client) - d.addErrback(self._changepassword_failed, client) + d.addCallback(client.with_msg_id(self._changepassword_done), client) + d.addErrback(client.with_msg_id(self._changepassword_failed), client) def _changepassword_done(self, verdict, client): if client.session_id not in self._root.clients: @@ -3628,8 +3634,8 @@ def in_LISTBANS(self, client): # 3.1: list_bans() returns a plain list of dicts, so it defers cleanly; the callback only # Sends, so it needs no session bracket. d = self._root.defer_db(self.bandb.list_bans) - d.addCallback(self._listbans_done, client) - d.addErrback(self._listbans_failed, client) + d.addCallback(client.with_msg_id(self._listbans_done), client) + d.addErrback(client.with_msg_id(self._listbans_failed), client) def _listbans_done(self, banlist, client): if client.session_id not in self._root.clients: @@ -3651,8 +3657,8 @@ def in_LISTBLACKLIST(self, client): # send the blacklist of domains for email verification # 3.1: list_blacklist() returns a plain list of dicts; defer it, callback only Sends. d = self._root.defer_db(self.bandb.list_blacklist) - d.addCallback(self._listblacklist_done, client) - d.addErrback(self._listblacklist_failed, client) + d.addCallback(client.with_msg_id(self._listblacklist_done), client) + d.addErrback(client.with_msg_id(self._listblacklist_failed), client) def _listblacklist_done(self, blacklist, client): if client.session_id not in self._root.clients: @@ -3707,8 +3713,8 @@ def in_LISTMODS(self, client): # 3.1: the access guard stays on the reactor; list_mods() returns plain (admins, mods) # strings, so the query defers and the callback only Sends. d = self._root.defer_db(self.userdb.list_mods) - d.addCallback(self._listmods_done, client) - d.addErrback(self._listmods_failed, client) + d.addCallback(client.with_msg_id(self._listmods_done), client) + d.addErrback(client.with_msg_id(self._listmods_failed), client) def _listmods_done(self, mods_tuple, client): if client.session_id not in self._root.clients: @@ -4025,8 +4031,8 @@ def in_CHANGEEMAIL(self, client, newmail, verification_code=""): # one atomic DB unit; the callback (on the reactor) sets the new email in memory, invalidates # the 1.2 cache and replies. It does no reactor-side DB, so it needs no session bracket. d = self._root.defer_db(self.userdb.do_change_email, client.username, newmail) - d.addCallback(self._changeemail_done, client, newmail) - d.addErrback(self._changeemail_failed, client, newmail) + d.addCallback(client.with_msg_id(self._changeemail_done), client, newmail) + d.addErrback(client.with_msg_id(self._changeemail_failed), client, newmail) def _changeemail_done(self, verdict, client, newmail): if client.session_id not in self._root.clients: @@ -4092,8 +4098,8 @@ def in_RESETPASSWORD(self, client, email, verification_code): uid = recover_client.user_id raw, hashed = self.userdb.generate_password() d = self._root.defer_db(self.userdb.do_set_password, uid, hashed) - d.addCallback(self._resetpassword_done, client, raw, uid) - d.addErrback(self._resetpassword_failed, client) + d.addCallback(client.with_msg_id(self._resetpassword_done), client, raw, uid) + d.addErrback(client.with_msg_id(self._resetpassword_failed), client) def _resetpassword_done(self, verdict, client, raw_password, uid): if client.session_id not in self._root.clients: @@ -4151,8 +4157,8 @@ def in_RESETUSERPASSWORD(self, client, username, newmail=None): uid = recover_client.user_id raw, hashed = self.userdb.generate_password() d = self._root.defer_db(self.userdb.do_set_password, uid, hashed, add_email) - d.addCallback(self._resetuserpassword_done, client, raw, uid) - d.addErrback(self._resetuserpassword_failed, client) + d.addCallback(client.with_msg_id(self._resetuserpassword_done), client, raw, uid) + d.addErrback(client.with_msg_id(self._resetuserpassword_failed), client) def _resetuserpassword_done(self, verdict, client, raw_password, uid): if client.session_id not in self._root.clients: @@ -4195,8 +4201,8 @@ def in_DELETEACCOUNT(self, client, username): # snapshot. The callback only invalidates the cache + Sends, so it needs no session bracket. _raw, hashed = self.userdb.generate_password() d = self._root.defer_db(self.userdb.do_scrub_account, delete_client.user_id, hashed) - d.addCallback(self._deleteaccount_done, client, delete_client.username) - d.addErrback(self._deleteaccount_failed, client, delete_client.username) + d.addCallback(client.with_msg_id(self._deleteaccount_done), client, delete_client.username) + d.addErrback(client.with_msg_id(self._deleteaccount_failed), client, delete_client.username) def _deleteaccount_done(self, verdict, client, username): if client.session_id not in self._root.clients: diff --git a/tests/integration/msgidtest.py b/tests/integration/msgidtest.py new file mode 100644 index 00000000..b4338efc --- /dev/null +++ b/tests/integration/msgidtest.py @@ -0,0 +1,138 @@ +""" +End-to-end tier: a command's #id is on its reply and on nothing else (issue #60). + +The id was stored on the Client and never cleared, so every later line to that client carried +it until the next command. Replies sent from a deferred DB callback had the opposite problem, +carrying the id of whatever command the client had sent since, or none. + + unrelated line after '#5 PING', a SAIDPRIVATE from another player carries no id + deferred reply '#7 GETIP ' then '#8 PING' in one write: the GETIP reply is #7 + deferred, no next '#9 LISTBANS' then a bare 'PING' in one write: the LISTBANS reply is #9 + errors and refusals '#10 NOTACOMMAND' is answered with #10 on both lines + +Run: activate the venv, then python3 tests/integration/msgidtest.py +""" +import os as _os, sys as _sys +_sys.path[:0] = [_os.path.join(_os.path.dirname(__file__), _os.pardir, _os.pardir), + _os.path.join(_os.path.dirname(__file__), _os.pardir)] +from testenv import DB_KWARGS, HOST, PORT +import socket, hashlib, base64, time, sys +import pymysql + +TAG = sys.argv[1] if len(sys.argv) > 1 else "m1" +PW = base64.b64encode(hashlib.md5(b"secretpw").digest()).decode() +MOD = "mid_%s_mod" % TAG +TALKER = "mid_%s_talk" % TAG +OFFLINE = "mid_%s_off" % TAG + +errors = [] +def check(cond, label): + if not cond: errors.append(label) + +def recv_until(s, substr, timeout=8): + s.settimeout(timeout); buf = b"" + try: + while substr.encode() not in buf: + chunk = s.recv(8192) + if not chunk: break + buf += chunk + except socket.timeout: pass + return buf.decode(errors="replace") + +def read_for(s, seconds): + s.settimeout(seconds); buf = b"" + try: + while True: + chunk = s.recv(8192) + if not chunk: break + buf += chunk + except socket.timeout: pass + return [ln for ln in buf.decode(errors="replace").splitlines() if ln.strip()] + +def connect(): + s = socket.create_connection((HOST, PORT)); recv_until(s, "\n"); return s + +def register_and_confirm(user): + s = connect() + s.sendall(("REGISTER %s %s\n" % (user, PW)).encode()); recv_until(s, "\n") + time.sleep(2.5) + s.sendall(("LOGIN %s %s 0 * TestClient\n" % (user, PW)).encode()); recv_until(s, "AGREEMENTEND") + s.sendall(b"CONFIRMAGREEMENT\n") + ok = "LOGININFOEND" in recv_until(s, "LOGININFOEND"); s.close(); return ok + +def login_keep(user): + s = connect() + s.sendall(("LOGIN %s %s 0 * TestClient\n" % (user, PW)).encode()) + resp = recv_until(s, "LOGININFOEND") + if "ACCEPTED" not in resp: + s.close(); return None + read_for(s, 0.5) # drain the rest of the login burst + return s + +def exchange(s, lines, seconds=1.5): + """Send several commands in one write, so the server reads them back to back.""" + s.sendall(("".join(ln + "\n" for ln in lines)).encode()) + got = read_for(s, seconds) + print("%-44r -> %r" % (lines, got)) + return got + + +for u in (MOD, TALKER, OFFLINE): + if not register_and_confirm(u): + print("ABORT: could not register %s" % u); sys.exit(1) + +conn = pymysql.connect(**DB_KWARGS); cur = conn.cursor() +cur.execute("UPDATE users SET access='mod' WHERE username=%s", (MOD,)) +cur.execute("SELECT last_ip FROM users WHERE username=%s", (OFFLINE,)); OFFLINE_IP = cur.fetchone()[0] +conn.close() + +mod = login_keep(MOD) +talker = login_keep(TALKER) +if not (mod and talker): + print("ABORT: test users could not log in"); sys.exit(1) +read_for(mod, 0.5) # the ADDUSER for TALKER, who logged in after MOD + +# the reply to an id'd command carries the id +got = exchange(mod, ["#5 PING"]) +check(got == ["#5 PONG"], "'#5 PING' should be answered '#5 PONG', got %r" % got) + +# a line that has nothing to do with that command does not +talker.sendall(("SAYPRIVATE %s hello there\n" % MOD).encode()) +got = read_for(mod, 1.0) +print("%-44r -> %r" % ("(SAYPRIVATE from %s)" % TALKER, got)) +check("SAIDPRIVATE %s hello there" % TALKER in got, + "a private message from another player should arrive with no id, got %r" % got) + +# a deferred reply carries the id of the command that asked, not of a later one +got = exchange(mod, ["#7 GETIP %s" % OFFLINE, "#8 PING"]) +check("#8 PONG" in got, "'#8 PING' should be answered '#8 PONG', got %r" % got) +check("#7 SERVERMSG <%s> was recently bound to %s" % (OFFLINE, OFFLINE_IP) in got, + "the deferred GETIP reply should carry #7, got %r" % got) + +# and still carries it when the next command had no id at all +got = exchange(mod, ["#9 LISTBANS", "PING"]) +check("PONG" in got, "a bare PING should be answered with a bare PONG, got %r" % got) +check("#9 SERVERMSG Banlist is empty" in got, + "the deferred LISTBANS reply should carry #9, got %r" % got) + +# refusals from _handle itself carry the id on both lines +got = exchange(mod, ["#10 NOTACOMMAND"]) +check(len(got) == 2 and all(ln.startswith("#10 ") for ln in got), + "both lines refusing '#10 NOTACOMMAND' should carry #10, got %r" % got) + +# after all of that, an unrelated line is still clean +talker.sendall(("SAYPRIVATE %s second\n" % MOD).encode()) +got = read_for(mod, 1.0) +print("%-44r -> %r" % ("(SAYPRIVATE from %s)" % TALKER, got)) +check("SAIDPRIVATE %s second" % TALKER in got, + "a later private message should still arrive with no id, got %r" % got) + +mod.close(); talker.close() + + +if errors: + print("FAIL (%d):" % len(errors)) + for e in errors: print(" -", e) + sys.exit(1) +print("RESULT: PASS") +sys.exit(0)