From 95971f40a7383450c354f4bcf34da96abaf6c663 Mon Sep 17 00:00:00 2001 From: Norbert Heusser Date: Wed, 6 Mar 2024 11:57:49 +0000 Subject: [PATCH 1/6] Improved python test print packet function by adding mqtt5 property printer Signed-off-by: Norbert Heusser --- test/mosq_test.py | 12 ++-- test/mqtt5_props.py | 163 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 169 insertions(+), 6 deletions(-) diff --git a/test/mosq_test.py b/test/mosq_test.py index e961d9b5..e7380e40 100644 --- a/test/mosq_test.py +++ b/test/mosq_test.py @@ -408,12 +408,12 @@ def to_string(packet): return s elif cmd == 0x20: # CONNACK - if len(packet) == 4: - (cmd, rl, resv, rc) = struct.unpack('!BBBB', packet) - return "CONNACK, rl="+str(rl)+", res="+str(resv)+", rc="+str(rc) - elif len(packet) == 5: - (cmd, rl, flags, reason_code, proplen) = struct.unpack('!BBBBB', packet) - return "CONNACK, rl="+str(rl)+", flags="+str(flags)+", rc="+str(reason_code)+", proplen="+str(proplen) + if len(packet) >= 4: + (cmd, rl, flags, reason_code) = struct.unpack('!BBBB', packet[0:4]) + s=f"CONNACK, rl={rl}, res/flags={flags}, rc={reason_code}" + if len(packet) > 4: + s = s+ f", properties={mqtt5_props.print_properties(packet[4:])}" + return s else: return "CONNACK, (not decoded)" diff --git a/test/mqtt5_props.py b/test/mqtt5_props.py index 9f6ad4bc..7ddd11c2 100644 --- a/test/mqtt5_props.py +++ b/test/mqtt5_props.py @@ -68,6 +68,169 @@ def pack_varint(varint): if varint == 0: return s + def prop_finalise(props): return pack_varint(len(props)) + props + +def gen_properties(properties_dict: list[dict]): + props = b"" + if properties_dict is None: + return props + for prop in properties_dict: + id = prop.get("identifier") + value = prop.get("value") + if id in ( + PROP_PAYLOAD_FORMAT_INDICATOR, + PROP_REQUEST_PROBLEM_INFO, + PROP_REQUEST_RESPONSE_INFO, + PROP_MAXIMUM_QOS, + PROP_RETAIN_AVAILABLE, + PROP_WILDCARD_SUB_AVAILABLE, + PROP_SUBSCRIPTION_ID_AVAILABLE, + PROP_SHARED_SUB_AVAILABLE, + ): + props += gen_byte_prop(id, value) + elif id in ( + PROP_MESSAGE_EXPIRY_INTERVAL, + PROP_SESSION_EXPIRY_INTERVAL, + PROP_WILL_DELAY_INTERVAL, + PROP_MAXIMUM_PACKET_SIZE, + ): + props += gen_uint32_prop(id, value) + elif id in ( + PROP_CONTENT_TYPE, + PROP_RESPONSE_TOPIC, + PROP_CORRELATION_DATA, + PROP_ASSIGNED_CLIENT_IDENTIFIER, + PROP_AUTHENTICATION_METHOD, + PROP_AUTHENTICATION_DATA, + PROP_RESPONSE_INFO, + PROP_SERVER_REFERENCE, + PROP_REASON_STRING, + ): + props += gen_string_prop(id, value) + elif id == PROP_SUBSCRIPTION_IDENTIFIER: + props += gen_varint_prop(id, value) + elif id in ( + PROP_SERVER_KEEP_ALIVE, + PROP_RECEIVE_MAXIMUM, + PROP_TOPIC_ALIAS_MAXIMUM, + PROP_TOPIC_ALIAS, + ): + props += gen_uint16_prop(id, value) + elif id == PROP_USER_PROPERTY: + props += gen_string_pair_prop(id, prop["name"], value) + return props + + +def unpack_varint(b: bytes): + def decode_len(b: bytes): + for i in range(len(b)): + if b[i] & 0x80 == 0: + return i + 1 + return 0 + + var_len = decode_len(b) + variant = 0 + for i in range(var_len - 1, -1, -1): + variant = 0x80 * variant + (struct.unpack("!B", b[i : i + 1])[0] & 0x7F) + return variant, var_len + + +def unpack_string(b: bytes): + str_len = struct.unpack("!B", b[0:1])[0] + str_value = struct.unpack(f"!{str_len}s", b[1 : str_len + 1])[0] + return str_value, str_len + 1 + + +def unpack_property(b: bytes): + id = struct.unpack("!B", b[0:1])[0] + if id == PROP_PAYLOAD_FORMAT_INDICATOR: + return "PAYLOAD_FORMAT_INDICATOR", struct.unpack("!B", b[1:2])[0], 2 + elif id == PROP_PAYLOAD_FORMAT_INDICATOR: + return "PAYLOAD_FORMAT_INDICATOR", struct.unpack("!B", b[1:2])[0], 2 + elif id == PROP_REQUEST_PROBLEM_INFO: + return "REQUEST_PROBLEM_INFO", struct.unpack("!B", b[1:2])[0], 2 + elif id == PROP_REQUEST_RESPONSE_INFO: + return "REQUEST_RESPONSE_INFO", struct.unpack("!B", b[1:2])[0], 2 + elif id == PROP_MAXIMUM_QOS: + return "MAXIMUM_QOS", struct.unpack("!B", b[1:2])[0], 2 + elif id == PROP_RETAIN_AVAILABLE: + return "RETAIN_AVAILABLE", struct.unpack("!B", b[1:2])[0], 2 + elif id == PROP_WILDCARD_SUB_AVAILABLE: + return "WILDCARD_SUB_AVAILABLE", struct.unpack("!B", b[1:2])[0], 2 + elif id == PROP_SUBSCRIPTION_ID_AVAILABLE: + return "SUBSCRIPTION_ID_AVAILABLE", struct.unpack("!B", b[1:2])[0], 2 + elif id == PROP_SHARED_SUB_AVAILABLE: + return "SHARED_SUB_AVAILABLE", struct.unpack("!B", b[1:2])[0], 2 + elif id == PROP_SERVER_KEEP_ALIVE: + return "SERVER_KEEP_ALIVE", struct.unpack("!H", b[1:3])[0], 3 + elif id == PROP_RECEIVE_MAXIMUM: + return "RECEIVE_MAXIMUM", struct.unpack("!H", b[1:3])[0], 3 + elif id == PROP_TOPIC_ALIAS_MAXIMUM: + return "TOPIC_ALIAS_MAXIMUM", struct.unpack("!H", b[1:3])[0], 3 + elif id == PROP_TOPIC_ALIAS: + return "TOPIC_ALIAS", struct.unpack("!H", b[1:3])[0], 3 + elif id == PROP_MESSAGE_EXPIRY_INTERVAL: + return "MESSAGE_EXPIRY_INTERVAL", struct.unpack("!B", b[1:5])[0], 5 + elif id == PROP_SESSION_EXPIRY_INTERVAL: + return "SESSION_EXPIRY_INTERVAL", struct.unpack("!B", b[1:5])[0], 5 + elif id == PROP_WILL_DELAY_INTERVAL: + return "WILL_DELAY_INTERVAL", struct.unpack("!B", b[1:5])[0], 5 + elif id == PROP_MAXIMUM_PACKET_SIZE: + return "MAXIMUM_PACKET_SIZE", struct.unpack("!B", b[1:5])[0], 5 + elif id == PROP_CONTENT_TYPE: + value, pack_len = unpack_string(b[1:]) + return "CONTENT_TYPE", value, pack_len + 1 + elif id == PROP_RESPONSE_TOPIC: + value, pack_len = unpack_string(b[1:]) + return "RESPONSE_TOPIC", value, pack_len + 1 + elif id == PROP_CORRELATION_DATA: + value, pack_len = unpack_string(b[1:]) + return "CORRELATION_DATA", value, pack_len + 1 + elif id == PROP_ASSIGNED_CLIENT_IDENTIFIER: + value, pack_len = unpack_string(b[1:]) + return "ASSIGNED_CLIENT_IDENTIFIER", value, pack_len + 1 + elif id == PROP_AUTHENTICATION_METHOD: + value, pack_len = unpack_string(b[1:]) + return "AUTHENTICATION_METHOD", value, pack_len + 1 + elif id == PROP_AUTHENTICATION_DATA: + value, pack_len = unpack_string(b[1:]) + return "AUTHENTICATION_DATA", value, pack_len + 1 + elif id == PROP_RESPONSE_INFO: + value, pack_len = unpack_string(b[1:]) + return "RESPONSE_INFO", value, pack_len + 1 + elif id == PROP_SERVER_REFERENCE: + value, pack_len = unpack_string(b[1:]) + return "SERVER_REFERENCE", value, pack_len + 1 + elif id == PROP_REASON_STRING: + value, pack_len = unpack_string(b[1:]) + return "REASON_STRING", value, pack_len + 1 + elif id == PROP_SUBSCRIPTION_IDENTIFIER: + value, pack_len = unpack_varint(b[1:]) + return "REASON_STRING", value, pack_len + 1 + elif id == PROP_USER_PROPERTY: + name, pack_name_len = unpack_string(b[1:]) + value, pack_value_len = unpack_varint(b[1 + pack_name_len :]) + return ( + f"PROP_USER_PROPERTY:{name}", + value, + 1 + pack_name_len + pack_value_len + pack_len, + ) + else: + return f"", f"not decoded (len<={len(b)-1})", len(b) + + +def print_properties(b: bytes): + _, offset = unpack_varint(b) + props = [] + while offset < len(b): + try: + key, value, prop_len = unpack_property(b[offset:]) + offset += prop_len + props.append(f"{key}:{value}") + except struct.error: + props.append(f"decode error at offset {offset}: {b}") + break + return f"[{','.join(props)}]" From 5c4505498e3289a00ee1386ff623da892169bdcc Mon Sep 17 00:00:00 2001 From: Norbert Heusser Date: Wed, 6 Mar 2024 12:04:50 +0000 Subject: [PATCH 2/6] Fixed handling of expired messages after restore from persistent plugin Signed-off-by: Norbert Heusser --- src/database.c | 14 +- src/plugin_public.c | 21 +- ...15-persist-client-drop-expired-messages.py | 197 ++++++++++++++++++ test/broker/Makefile | 1 + test/broker/persist_sqlite.py | 17 ++ test/unit/broker/persist_read_stubs.c | 83 ++++---- 6 files changed, 272 insertions(+), 61 deletions(-) create mode 100755 test/broker/15-persist-client-drop-expired-messages.py diff --git a/src/database.c b/src/database.c index 3a39c95f..89f386e1 100644 --- a/src/database.c +++ b/src/database.c @@ -380,6 +380,10 @@ static void db__fill_inflight_out_from_queue(struct mosquitto *context) client_msg->data.state = mosq_ms_publish_qos2; break; } + if(client_msg->base_msg->data.expiry_time && db.now_real_s > client_msg->base_msg->data.expiry_time){ + db__message_remove_queued(context, &context->msgs_out, client_msg); + continue; + } plugin_persist__handle_client_msg_update(context, client_msg); db__message_dequeue_first(context, &context->msgs_out); } @@ -942,9 +946,11 @@ int db__message_store(const struct mosquitto *source, struct mosquitto__base_msg } base_msg->origin = origin; if(message_expiry_interval){ - base_msg->data.expiry_time = db.now_real_s + (*message_expiry_interval); - }else{ - base_msg->data.expiry_time = 0; + if(*message_expiry_interval > 0){ + base_msg->data.expiry_time = db.now_real_s + *message_expiry_interval; + }else{ + base_msg->data.expiry_time = 0; + } } base_msg->dest_ids = NULL; @@ -1207,6 +1213,7 @@ void db__expire_all_messages(struct mosquitto *context) db__message_remove_inflight(context, &context->msgs_out, client_msg); } } + db__fill_inflight_out_from_queue(context); DL_FOREACH_SAFE(context->msgs_out.queued, client_msg, tmp){ if(client_msg->base_msg->data.expiry_time && db.now_real_s > client_msg->base_msg->data.expiry_time){ db__message_remove_queued(context, &context->msgs_out, client_msg); @@ -1289,6 +1296,7 @@ static int db__message_write_inflight_out_single(struct mosquitto *context, stru util__increment_send_quota(context); } db__message_remove_inflight(context, &context->msgs_out, client_msg); + db__fill_inflight_out_from_queue(context); return MOSQ_ERR_SUCCESS; }else{ expiry_interval = (uint32_t)(base_msg->data.expiry_time - db.now_real_s); diff --git a/src/plugin_public.c b/src/plugin_public.c index 94c3e7c7..52b937f7 100644 --- a/src/plugin_public.c +++ b/src/plugin_public.c @@ -750,9 +750,7 @@ BROKER_EXPORT int mosquitto_persist_base_msg_add(struct mosquitto_base_msg *msg_ { struct mosquitto context; struct mosquitto__base_msg *base_msg; - uint32_t message_expiry_interval; - uint32_t *p_message_expiry_interval; - time_t message_expiry_interval_tt; + int i; int rc; memset(&context, 0, sizeof(context)); @@ -766,25 +764,12 @@ BROKER_EXPORT int mosquitto_persist_base_msg_add(struct mosquitto_base_msg *msg_ context.id = (char *)msg_add->source_id; context.username = (char *)msg_add->source_username; - p_message_expiry_interval = &message_expiry_interval; - if(msg_add->expiry_time == 0){ - p_message_expiry_interval = NULL; - }else if(msg_add->expiry_time <= db.now_real_s){ - message_expiry_interval = 0; - }else{ - message_expiry_interval_tt = msg_add->expiry_time - db.now_real_s; - if(message_expiry_interval_tt > UINT32_MAX){ - message_expiry_interval = UINT32_MAX; - }else{ - message_expiry_interval = (uint32_t)message_expiry_interval_tt; - } - } - base_msg = mosquitto_calloc(1, sizeof(struct mosquitto__base_msg)); if(base_msg == NULL){ goto error; } base_msg->data.store_id = msg_add->store_id; + base_msg->data.expiry_time = msg_add->expiry_time; base_msg->data.payloadlen = msg_add->payloadlen; base_msg->data.source_mid = msg_add->source_mid; base_msg->data.qos = msg_add->qos; @@ -807,7 +792,7 @@ BROKER_EXPORT int mosquitto_persist_base_msg_add(struct mosquitto_base_msg *msg_ } base_msg->stored = true; - rc = db__message_store(&context, base_msg, p_message_expiry_interval, mosq_mo_broker); + rc = db__message_store(&context, base_msg, NULL, mosq_mo_broker); return rc; error: diff --git a/test/broker/15-persist-client-drop-expired-messages.py b/test/broker/15-persist-client-drop-expired-messages.py new file mode 100755 index 00000000..48a37214 --- /dev/null +++ b/test/broker/15-persist-client-drop-expired-messages.py @@ -0,0 +1,197 @@ +#!/usr/bin/env python3 + +# Connect a client, add a subscription, disconnect, send a message with a +# different client, restore, reconnect, check it is received. + +from mosq_test_helper import * + +persist_help = persist_module() + +port = mosq_test.get_port() + +num_messages = 100 + +def do_test(test_case_name: str, additional_config_entries: dict): + conf_file = os.path.basename(__file__).replace(".py", f"_{port}.conf") + persist_help.write_config( + conf_file, + port, + additional_config_entries=additional_config_entries, + ) + persist_help.init(port) + + client_id = "test-expired-messages-subscriber" + username = "test-message-expiry" + + qos = 1 + topic = "client-msg/test" + source_id = "test-expired-messages-publisher" + proto_ver = 5 + + connect_packet = mosq_test.gen_connect( + client_id, username=username, proto_ver=proto_ver, clean_session=False, session_expiry=60 + ) + connack_packet1 = mosq_test.gen_connack(rc=0, proto_ver=proto_ver) + connack_packet2 = mosq_test.gen_connack(rc=0, flags=1, proto_ver=proto_ver) + + mid = 1 + subscribe_packet = mosq_test.gen_subscribe(mid, topic, qos, proto_ver=proto_ver) + suback_packet = mosq_test.gen_suback(mid, qos=qos, proto_ver=proto_ver) + + connect2_packet = mosq_test.gen_connect( + source_id, username=username, proto_ver=proto_ver + ) + connack2_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver) + + rc = 1 + + broker = mosq_test.start_broker( + filename=conf_file, use_conf=True, port=port + ) + + con = None + try: + sock = mosq_test.do_client_connect( + connect_packet, connack_packet1, timeout=5, port=port + ) + mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback") + sock.close() + + sock = mosq_test.do_client_connect( + connect2_packet, connack2_packet, timeout=5, port=port + ) + props = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_MESSAGE_EXPIRY_INTERVAL, 60) + for i in range(num_messages): + payload = f"queued message {i:3}" + mid = 10 + i + publish_packet = mosq_test.gen_publish( + topic, + mid=mid, + qos=qos, + payload=payload.encode("UTF-8"), + proto_ver=proto_ver, + properties=props, + ) + puback_packet = mosq_test.gen_puback(mid=mid, proto_ver=proto_ver) + mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback") + sock.close() + + # Terminate the broker + (broker_terminate_rc, stde) = mosq_test.terminate_broker(broker) + broker = None + + persist_help.check_counts( + port, + clients=1, + client_msgs_out=num_messages, + base_msgs=num_messages, + subscriptions=1, + ) + + # Check client + # persist_help.check_client( + # port, + # client_id, + # username=username, + # will_delay_time=0, + # session_expiry_time=60, + # listener_port=port, + # max_packet_size=0, + # max_qos=2, + # retain_available=1, + # session_expiry_interval=60, + # will_delay_interval=0, + # ) + + # Check subscription + persist_help.check_subscription(port, client_id, topic, qos, 0) + + # Check stored message + for i in range(num_messages): + payload = f"queued message {i:3}" + payload_b = payload.encode("UTF-8") + mid = 10 + i + store_id = persist_help.check_base_msg( + port, + 60, + topic, + payload_b, + source_id, + username, + len(payload_b), + mid, + port, + qos, + retain=0, + idx=i, + ) + + # Check client msg + subscriber_mid = 1 + i + cmsg_id = 1 + i + persist_help.check_client_msg( + port, + client_id, + cmsg_id, + store_id, + 0, + persist_help.dir_out, + subscriber_mid, + qos, + 0, + persist_help.ms_queued, + idx=i, + ) + + # Modify message expiry timestamp in the database + assert persist_help.modify_base_msgs(port, sub_expiry_time=120) == num_messages + + # Restart broker + broker = mosq_test.start_broker( + filename=conf_file, use_conf=True, port=port + ) + + # Connect client again, it should have a session, but all queued messages should be dropped + sock = mosq_test.do_client_connect( + connect_packet, connack_packet2, timeout=5, port=port, + ) + + # Send ping and wait for the PINGRESP to make sure the broker will not send a queued message instead + mosq_test.do_ping(sock) + sock.close() + + (broker_terminate_rc, stde) = mosq_test.terminate_broker(broker) + broker = None + + persist_help.check_counts( + port, + clients=1, + client_msgs_out=0, + base_msgs=0, + subscriptions=1, + ) + + rc = broker_terminate_rc + finally: + if broker is not None: + broker.terminate() + if mosq_test.wait_for_subprocess(broker): + if rc == 0: + rc = 1 + (_, stde) = broker.communicate() + os.remove(conf_file) + rc += persist_help.cleanup(port) + + print(f"{test_case_name}") + if rc: + print(stde.decode("utf-8")) + assert rc == 0, f"rc: {rc}" + + +do_test( + "memory queue", + additional_config_entries={ + "log_type": "all", + "max_queued_messages": num_messages, + }, +) diff --git a/test/broker/Makefile b/test/broker/Makefile index c6b1f764..ef4cd3ba 100644 --- a/test/broker/Makefile +++ b/test/broker/Makefile @@ -273,6 +273,7 @@ endif PERSIST_TESTS = \ ./15-persist-bridge-queue.py \ + ./15-persist-client-drop-expired-messages.py \ ./15-persist-client-msg-in-v3-1-1.py \ ./15-persist-client-msg-in-v5-0.py \ ./15-persist-client-msg-modify-acl.py \ diff --git a/test/broker/persist_sqlite.py b/test/broker/persist_sqlite.py index e0bd606a..8a2c78be 100755 --- a/test/broker/persist_sqlite.py +++ b/test/broker/persist_sqlite.py @@ -377,6 +377,23 @@ def check_base_msg( return row[0] +def modify_base_msgs( + port : int, + sub_expiry_time : int +): + num_modified_rows = 0 + con = sqlite3.connect(f"{port}/mosquitto.sqlite3") + try: + cur = con.cursor() + cur.execute( + "UPDATE base_msgs" + +f" SET expiry_time = expiry_time - {sub_expiry_time}" + ) + num_modified_rows = cur.rowcount + con.commit() + finally: + con.close() + return num_modified_rows def check_retain(port, topic, store_id): con = sqlite3.connect(f"{port}/mosquitto.sqlite3") diff --git a/test/unit/broker/persist_read_stubs.c b/test/unit/broker/persist_read_stubs.c index 76d89733..d84147a7 100644 --- a/test/unit/broker/persist_read_stubs.c +++ b/test/unit/broker/persist_read_stubs.c @@ -46,51 +46,54 @@ void db__msg_store_free(struct mosquitto__base_msg *store) int db__message_store(const struct mosquitto *source, struct mosquitto__base_msg *stored, uint32_t *message_expiry_interval, enum mosquitto_msg_origin origin) { - int rc = MOSQ_ERR_SUCCESS; - + int rc = MOSQ_ERR_SUCCESS; + UNUSED(origin); - - if(source && source->id){ - stored->data.source_id = mosquitto__strdup(source->id); - }else{ - stored->data.source_id = mosquitto__strdup(""); - } - if(!stored->data.source_id){ - rc = MOSQ_ERR_NOMEM; - goto error; - } - - if(source && source->username){ - stored->data.source_username = mosquitto__strdup(source->username); - if(!stored->data.source_username){ - rc = MOSQ_ERR_NOMEM; - goto error; - } - } - if(source){ + + if(source && source->id){ + stored->data.source_id = mosquitto__strdup(source->id); + }else{ + stored->data.source_id = mosquitto__strdup(""); + } + if(!stored->data.source_id){ + rc = MOSQ_ERR_NOMEM; + goto error; + } + + if(source && source->username){ + stored->data.source_username = mosquitto__strdup(source->username); + if(!stored->data.source_username){ + rc = MOSQ_ERR_NOMEM; + goto error; + } + } + if(source){ stored->source_listener = source->listener; - } - if(message_expiry_interval){ - stored->data.expiry_time = time(NULL) + (*message_expiry_interval); - }else{ - stored->data.expiry_time = 0; - } - - stored->dest_ids = NULL; - stored->dest_id_count = 0; - db.msg_store_count++; - db.msg_store_bytes += stored->data.payloadlen; - - if(!stored->data.store_id){ - stored->data.store_id = ++db.last_db_id; - } - + } + + if (message_expiry_interval){ + if(*message_expiry_interval > 0){ + stored->data.expiry_time = time(NULL) + *message_expiry_interval; + }else{ + stored->data.expiry_time = 0; + } + } + + stored->dest_ids = NULL; + stored->dest_id_count = 0; + db.msg_store_count++; + db.msg_store_bytes += stored->data.payloadlen; + + if(!stored->data.store_id){ + stored->data.store_id = ++db.last_db_id; + } + HASH_ADD(hh, db.msg_store, data.store_id, sizeof(stored->data.store_id), stored); - - return MOSQ_ERR_SUCCESS; + + return MOSQ_ERR_SUCCESS; error: db__msg_store_free(stored); - return rc; + return rc; } int log__printf(struct mosquitto *mosq, unsigned int priority, const char *fmt, ...) From b8564daa20c39569ec990596ae24f034e4b7bf29 Mon Sep 17 00:00:00 2001 From: Norbert Heusser Date: Thu, 7 Mar 2024 20:27:25 +0000 Subject: [PATCH 3/6] Fixed deletion of expired session after restore from persistence plugin Signed-off-by: Norbert Heusser --- plugins/persist-sqlite/base_msgs.c | 18 ++ plugins/persist-sqlite/clients.c | 7 +- plugins/persist-sqlite/init.c | 24 +- plugins/persist-sqlite/persist_sqlite.h | 2 + src/mosquitto.c | 1 + src/plugin_public.c | 1 + src/session_expiry.c | 4 +- .../15-persist-client-expired-session.py | 216 ++++++++++++++++++ test/broker/Makefile | 1 + test/broker/persist_sqlite.py | 25 +- 10 files changed, 293 insertions(+), 6 deletions(-) create mode 100755 test/broker/15-persist-client-expired-session.py diff --git a/plugins/persist-sqlite/base_msgs.c b/plugins/persist-sqlite/base_msgs.c index e1d40189..1b04a496 100644 --- a/plugins/persist-sqlite/base_msgs.c +++ b/plugins/persist-sqlite/base_msgs.c @@ -215,3 +215,21 @@ int persist_sqlite__base_msg_remove_cb(int event, void *event_data, void *userda return rc; } + +int persist_sqlite__base_msg_clear(struct mosquitto_sqlite *ms, const char *clientid) +{ + int rc = MOSQ_ERR_UNKNOWN; + + if(sqlite3_bind_text(ms->base_msg_remove_for_clientid_stmt, 1, clientid, (int)strlen(clientid), SQLITE_STATIC) == SQLITE_OK){ + ms->event_count++; + rc = sqlite3_step(ms->base_msg_remove_for_clientid_stmt); + if(rc == SQLITE_DONE){ + rc = MOSQ_ERR_SUCCESS; + }else{ + rc = MOSQ_ERR_UNKNOWN; + } + } + sqlite3_reset(ms->base_msg_remove_for_clientid_stmt); + + return rc; +} diff --git a/plugins/persist-sqlite/clients.c b/plugins/persist-sqlite/clients.c index a0d2d84a..5e3f2073 100644 --- a/plugins/persist-sqlite/clients.c +++ b/plugins/persist-sqlite/clients.c @@ -89,6 +89,11 @@ int persist_sqlite__client_remove_cb(int event, void *event_data, void *userdata rc = MOSQ_ERR_UNKNOWN; } } + + /* Delete base msgs before deletion of client_msgs as the query will iterate over the client_msgs table */ + persist_sqlite__base_msg_clear(ms, ed->data.clientid); + persist_sqlite__client_msg_clear(ms, ed->data.clientid); + if(sqlite3_bind_text(ms->client_remove_stmt, 1, ed->data.clientid, (int)strlen(ed->data.clientid), SQLITE_STATIC) == SQLITE_OK){ @@ -101,7 +106,6 @@ int persist_sqlite__client_remove_cb(int event, void *event_data, void *userdata rc = MOSQ_ERR_UNKNOWN; } } - persist_sqlite__client_msg_clear(ms, ed->data.clientid); return rc; } @@ -133,3 +137,4 @@ int persist_sqlite__client_update_cb(int event, void *event_data, void *userdata return rc; } + diff --git a/plugins/persist-sqlite/init.c b/plugins/persist-sqlite/init.c index 93ca8c2e..bb2c515c 100644 --- a/plugins/persist-sqlite/init.c +++ b/plugins/persist-sqlite/init.c @@ -141,8 +141,19 @@ static int create_tables(struct mosquitto_sqlite *ms) "CREATE INDEX IF NOT EXISTS client_msgs_client_id ON client_msgs(client_id);", NULL, NULL, NULL); if(rc) goto fail; + rc = sqlite3_exec(ms->db, - "CREATE INDEX IF NOT EXISTS client_msgs_store_id ON client_msgs(store_id);", + "DROP INDEX IF EXISTS client_msgs_store_id;", + NULL, NULL, NULL); + if(rc) goto fail; + + rc = sqlite3_exec(ms->db, + "CREATE INDEX IF NOT EXISTS client_msgs_store_id ON client_msgs(store_id,client_id);", + NULL, NULL, NULL); + if(rc) goto fail; + + rc = sqlite3_exec(ms->db, + "CREATE INDEX IF NOT EXISTS retains_storeid ON retains(store_id);", NULL, NULL, NULL); if(rc) goto fail; @@ -294,6 +305,17 @@ static int prepare_statements(struct mosquitto_sqlite *ms) &ms->base_msg_remove_stmt, NULL); if(rc) goto fail; + rc = sqlite3_prepare_v3(ms->db, + "DELETE FROM base_msgs AS bm " + "WHERE bm.store_id IN " + "( SELECT cm.store_id FROM client_msgs AS cm" + " LEFT OUTER JOIN client_msgs AS oc ON oc.store_id = cm.store_id AND oc.client_id != cm.client_id" + " LEFT OUTER JOIN retains AS rm ON rm.store_id = cm.store_id" + " WHERE cm.client_id = ? AND oc.store_id IS NULL AND rm.store_id IS NULL)", + -1, SQLITE_PREPARE_PERSISTENT, + &ms->base_msg_remove_for_clientid_stmt, NULL); + if(rc) goto fail; + rc = sqlite3_prepare_v3(ms->db, "SELECT store_id, expiry_time, topic, payload, source_id, source_username, " "payloadlen, source_mid, source_port, qos, retain, properties " diff --git a/plugins/persist-sqlite/persist_sqlite.h b/plugins/persist-sqlite/persist_sqlite.h index ad629df2..6c0e7d88 100644 --- a/plugins/persist-sqlite/persist_sqlite.h +++ b/plugins/persist-sqlite/persist_sqlite.h @@ -43,6 +43,7 @@ struct mosquitto_sqlite { sqlite3_stmt *client_msg_clear_all_stmt; sqlite3_stmt *base_msg_add_stmt; sqlite3_stmt *base_msg_remove_stmt; + sqlite3_stmt *base_msg_remove_for_clientid_stmt; sqlite3_stmt *base_msg_load_stmt; sqlite3_stmt *retain_msg_set_stmt; sqlite3_stmt *retain_msg_remove_stmt; @@ -70,6 +71,7 @@ int persist_sqlite__client_msg_update_cb(int event, void *event_data, void *user int persist_sqlite__base_msg_add_cb(int event, void *event_data, void *userdata); int persist_sqlite__base_msg_load_cb(int event, void *event_data, void *userdata); int persist_sqlite__base_msg_remove_cb(int event, void *event_data, void *userdata); +int persist_sqlite__base_msg_clear(struct mosquitto_sqlite *ms, const char *clientid); int persist_sqlite__retain_msg_set_cb(int event, void *event_data, void *userdata); int persist_sqlite__retain_msg_remove_cb(int event, void *event_data, void *userdata); int persist_sqlite__subscription_add_cb(int event, void *event_data, void *userdata); diff --git a/src/mosquitto.c b/src/mosquitto.c index f4e415a2..210e72ef 100644 --- a/src/mosquitto.c +++ b/src/mosquitto.c @@ -426,6 +426,7 @@ int main(int argc, char *argv[]) if(rc) return rc; plugin_persist__handle_restore(); + session_expiry__check(); db__msg_store_compact(); /* After loading persisted clients and ACLs, try to associate them, diff --git a/src/plugin_public.c b/src/plugin_public.c index 52b937f7..3d9da0db 100644 --- a/src/plugin_public.c +++ b/src/plugin_public.c @@ -517,6 +517,7 @@ BROKER_EXPORT int mosquitto_persist_client_add(struct mosquitto_client *client) } context__add_to_by_id(context); + session_expiry__add_from_persistence(context,context->session_expiry_time); return MOSQ_ERR_SUCCESS; error: diff --git a/src/session_expiry.c b/src/session_expiry.c index 138a0b2b..8dbc3d1c 100644 --- a/src/session_expiry.c +++ b/src/session_expiry.c @@ -151,9 +151,9 @@ void session_expiry__check(void) struct mosquitto *context; time_t timeout; - if(db.now_real_s <= last_check){ + if(last_check != 0 && db.now_real_s <= last_check){ if(expiry_list){ - /* Next event is the first item of the list, we must set the timeout even if we aren't + /* Next event is the first item of the list, we must set the timeout even if we aren't * checking the full list */ timeout = (expiry_list->context->session_expiry_time - db.now_real_s) * 1000; if(timeout <= 0){ diff --git a/test/broker/15-persist-client-expired-session.py b/test/broker/15-persist-client-expired-session.py new file mode 100755 index 00000000..d42d92b1 --- /dev/null +++ b/test/broker/15-persist-client-expired-session.py @@ -0,0 +1,216 @@ +#!/usr/bin/env python3 + +# Connect a client, add a subscription, disconnect, send a message with a +# different client, restore, reconnect, check it is received. + +from mosq_test_helper import * + +persist_help = persist_module() + +port = mosq_test.get_port() + +num_messages = 100 + + +def do_test( + test_case_name: str, additional_config_entries: dict, message_expiry_interval: int +): + conf_file = os.path.basename(__file__).replace(".py", f"_{port}.conf") + persist_help.write_config( + conf_file, + port, + additional_config_entries=additional_config_entries, + ) + persist_help.init(port) + + client_id = "test-expired-session-subscriber" + username = "test-session-expiry" + + qos = 1 + topic = "client-msg/test" + source_id = "test-expired-session-publisher" + proto_ver = 5 + + connect_packet = mosq_test.gen_connect( + client_id, + username=username, + proto_ver=proto_ver, + clean_session=False, + session_expiry=60, + ) + connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver) + + mid = 1 + subscribe_packet = mosq_test.gen_subscribe(mid, topic, qos, proto_ver=proto_ver) + suback_packet = mosq_test.gen_suback(mid, qos=qos, proto_ver=proto_ver) + + connect2_packet = mosq_test.gen_connect( + source_id, username=username, proto_ver=proto_ver + ) + + rc = 1 + + broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port) + + con = None + try: + sock = mosq_test.do_client_connect( + connect_packet, connack_packet, timeout=5, port=port + ) + mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback") + sock.close() + + sock = mosq_test.do_client_connect( + connect2_packet, connack_packet, timeout=5, port=port + ) + props = ( + mqtt5_props.gen_uint32_prop( + mqtt5_props.PROP_MESSAGE_EXPIRY_INTERVAL, message_expiry_interval + ) + if message_expiry_interval > 0 + else b"" + ) + for i in range(num_messages): + payload = f"queued message {i:3}" + mid = 10 + i + publish_packet = mosq_test.gen_publish( + topic, + mid=mid, + qos=qos, + payload=payload.encode("UTF-8"), + proto_ver=proto_ver, + properties=props, + ) + puback_packet = mosq_test.gen_puback(mid=mid, proto_ver=proto_ver) + mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback") + sock.close() + + # Terminate the broker + (broker_terminate_rc, stde) = mosq_test.terminate_broker(broker) + broker = None + + persist_help.check_counts( + port, + clients=1, + client_msgs_out=num_messages, + base_msgs=num_messages, + subscriptions=1, + ) + + # Check client + persist_help.check_client( + port, + client_id, + username=username, + will_delay_time=0, + session_expiry_time=60, + listener_port=port, + max_packet_size=0, + max_qos=2, + retain_available=1, + session_expiry_interval=60, + will_delay_interval=0, + ) + + # Check subscription + persist_help.check_subscription(port, client_id, topic, qos, 0) + + # Check stored message + for i in range(num_messages): + payload = f"queued message {i:3}" + payload_b = payload.encode("UTF-8") + mid = 10 + i + store_id = persist_help.check_base_msg( + port, + message_expiry_interval, + topic, + payload_b, + source_id, + username, + len(payload_b), + mid, + port, + qos, + retain=0, + idx=i, + ) + + # Check client msg + subscriber_mid = 1 + i + cmsg_id = 1 + i + persist_help.check_client_msg( + port, + client_id, + cmsg_id, + store_id, + 0, + persist_help.dir_out, + subscriber_mid, + qos, + 0, + persist_help.ms_queued, + idx=i, + ) + + # Put session expiry_time into the past + assert persist_help.modify_client(port, client_id, sub_expiry_time=120) == 1 + + # Restart broker + broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port) + + # Connect client again, it should have a session, but all queued messages should be dropped + sock = mosq_test.do_client_connect( + connect_packet, + connack_packet, + timeout=5, + port=port, + ) + + # Send ping and wait for the PINGRESP to make sure the broker will not send a queued message instead + mosq_test.do_ping(sock) + sock.close() + + (broker_terminate_rc, stde) = mosq_test.terminate_broker(broker) + broker = None + + persist_help.check_counts( + port, + clients=1, + client_msgs_out=0, + base_msgs=0, + subscriptions=0, + ) + + rc = broker_terminate_rc + finally: + if broker is not None: + broker.terminate() + if mosq_test.wait_for_subprocess(broker): + if rc == 0: + rc = 1 + (_, stde) = broker.communicate() + os.remove(conf_file) + rc += persist_help.cleanup(port) + + print(f"{test_case_name}") + if rc: + print(stde.decode("utf-8")) + assert rc == 0, f"rc: {rc}" + + +memory_queue_config = { + "log_type": "all", + "max_queued_messages": num_messages, +} + + +do_test( + "memory queue, message expiry interval: 0", + additional_config_entries=memory_queue_config, + message_expiry_interval=0, +) +do_test( + "memory queue, message expiry interval: 120", + additional_config_entries=memory_queue_config, + message_expiry_interval=120, +) diff --git a/test/broker/Makefile b/test/broker/Makefile index ef4cd3ba..3b80e90a 100644 --- a/test/broker/Makefile +++ b/test/broker/Makefile @@ -274,6 +274,7 @@ endif PERSIST_TESTS = \ ./15-persist-bridge-queue.py \ ./15-persist-client-drop-expired-messages.py \ + ./15-persist-client-expired-session.py \ ./15-persist-client-msg-in-v3-1-1.py \ ./15-persist-client-msg-in-v5-0.py \ ./15-persist-client-msg-modify-acl.py \ diff --git a/test/broker/persist_sqlite.py b/test/broker/persist_sqlite.py index 8a2c78be..9c5a7bc7 100755 --- a/test/broker/persist_sqlite.py +++ b/test/broker/persist_sqlite.py @@ -28,7 +28,7 @@ def write_config(filename, port, additional_config_entries : dict = {}): f"plugin {mosq_test.get_build_root()}/plugins/persist-sqlite/mosquitto_persist_sqlite.so\n" ) f.write("plugin_opt_db_file %d/mosquitto.sqlite3\n" % (port)) - for entry, value in additional_config_entries.items(): + for entry, value in additional_config_entries.items(): f.write(f"{entry} {value}\n") @@ -236,6 +236,27 @@ def check_client( ) con.close() +def modify_client( + port : int, + client_id : str, + sub_expiry_time : int +): + num_modified_rows = 0 + con = sqlite3.connect(f"{port}/mosquitto.sqlite3") + try: + cur = con.cursor() + cur.execute( + "UPDATE clients" + +f" SET session_expiry_time = session_expiry_time - {sub_expiry_time}" + +f" WHERE client_id = ?", + (client_id,) + ) + num_modified_rows = cur.rowcount + con.commit() + finally: + con.close() + + return num_modified_rows def check_subscription( port, client_id, topic, subscription_options, subscription_identifier @@ -337,7 +358,7 @@ def check_base_msg( ) for i in range(0, idx + 1): row = cur.fetchone() - + if row[0] == 0: raise ValueError("Invalid store_id %d / %d" % (row[0], store_id)) From 9b26357c173b4405f5ce38d159e2a8acea6d9474 Mon Sep 17 00:00:00 2001 From: Norbert Heusser Date: Mon, 11 Mar 2024 09:37:42 +0000 Subject: [PATCH 4/6] Reformatting and improvement of persistent module test helpers Signed-off-by: Norbert Heusser --- test/broker/15-persist-bridge-queue.py | 125 +++++++++++------- ...15-persist-client-drop-expired-messages.py | 27 ++-- .../15-persist-client-expired-session.py | 1 - .../15-persist-client-msg-modify-acl.py | 9 +- test/broker/mosq_test_helper.py | 54 +++++--- test/broker/persist_sqlite.py | 102 +++++++++----- 6 files changed, 206 insertions(+), 112 deletions(-) diff --git a/test/broker/15-persist-bridge-queue.py b/test/broker/15-persist-bridge-queue.py index 0598686a..0186cb5e 100755 --- a/test/broker/15-persist-bridge-queue.py +++ b/test/broker/15-persist-bridge-queue.py @@ -24,6 +24,7 @@ conf_file_bridge_target = os.path.basename(__file__).replace( ".py", "_bridge_target.conf" ) + def do_test(test_case_name: str, bridging_add_config: dict, target_add_config: dict): persist_help.write_config( conf_file, port, additional_config_entries=bridging_add_config @@ -46,6 +47,18 @@ def do_test(test_case_name: str, bridging_add_config: dict, target_add_config: d source_id = "persist-bridge-test-publisher" proto_ver = 4 + def gen_pub_packets(idx: int, mid_offset: int): + payload = f"queued message {idx:3}" + publish_packet = mosq_test.gen_publish( + topic, + mid=mid_offset + idx, + qos=qos, + payload=payload.encode("UTF-8"), + proto_ver=proto_ver, + ) + puback_packet = mosq_test.gen_puback(mid=mid_offset + idx, proto_ver=proto_ver) + return publish_packet, puback_packet + connect_packet = mosq_test.gen_connect( client_id, proto_ver=proto_ver, clean_session=False ) @@ -67,44 +80,56 @@ def do_test(test_case_name: str, bridging_add_config: dict, target_add_config: d bridge_target_broker = mosq_test.start_broker( filename=conf_file_bridge_target, use_conf=True, port=bridge_target_port ) - - # Connect and send a ping to make sure bridge target broker is up - sock = mosq_test.do_client_connect( + + # Connect to the bridge target broker and make a qos1 subscription + sock_bridge_target = mosq_test.do_client_connect( connect_packet, connack_packet1, timeout=5, port=bridge_target_port ) - mosq_test.do_ping(sock) - sock.close() + mosq_test.do_send_receive( + sock_bridge_target, + subscribe_packet, + suback_packet, + "suback from bridge target", + ) # Now start the broker with the bridge broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port) - # Connect and send a ping to make sure bridging broker is up + # Connect and send a single message forwarded to the bridge target sock = mosq_test.do_client_connect( connect2_packet, connack2_packet, timeout=5, port=port ) + publish_packet, puback_packet = gen_pub_packets(0, mid_offset=3) + mosq_test.do_send_receive( + sock, publish_packet, puback_packet, "puback for first message" + ) + + # Wait until we have received the message from the bridge target + publish_packet, puback_packet = gen_pub_packets(0, mid_offset=1) + mosq_test.do_receive_send( + sock_bridge_target, publish_packet, puback_packet, "first published message" + ) + + # Wait for a ping response to make sure the target broker has processed the PUBACK + mosq_test.do_ping(sock_bridge_target) + sock_bridge_target.close() + + # Make sure the bridging broker processes a ping, which means the PUBACK from the bridge target for the + # first message was processed as well mosq_test.do_ping(sock) # Stop the bridge target broker - (broker_terminate_rc, stde) = mosq_test.terminate_broker(bridge_target_broker) + (broker_terminate_rc, stde2) = mosq_test.terminate_broker(bridge_target_broker) bridge_target_broker = None # Publish messages for i in range(num_messages): - payload = f"queued message {i:3}" - mid = 10 + i - publish_packet = mosq_test.gen_publish( - topic, - mid=mid, - qos=qos, - payload=payload.encode("UTF-8"), - proto_ver=proto_ver, - ) - puback_packet = mosq_test.gen_puback(mid=mid, proto_ver=proto_ver) + publish_packet, puback_packet = gen_pub_packets(idx=i, mid_offset=10) mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback") sock.close() # Terminate the bridging broker - (broker_terminate_rc, stde) = mosq_test.terminate_broker(broker) + (broker_terminate_rc, stde3) = mosq_test.terminate_broker(broker) broker = None persist_help.check_counts( @@ -136,8 +161,8 @@ def do_test(test_case_name: str, bridging_add_config: dict, target_add_config: d ) # Check client msg - subscriber_mid = 3 + i - cmsg_id = 1 + i + subscriber_mid = 4 + i + cmsg_id = 2 + i persist_help.check_client_msg( port, "upstream-bridge", @@ -149,7 +174,6 @@ def do_test(test_case_name: str, bridging_add_config: dict, target_add_config: d qos, 0, persist_help.ms_queued, - idx=i ) # Start the bridge target broker @@ -157,27 +181,17 @@ def do_test(test_case_name: str, bridging_add_config: dict, target_add_config: d filename=conf_file_bridge_target, use_conf=True, port=bridge_target_port ) - # Connect to the bridge target broker and make a qos1 subscription + # Reconnect to the bridge target broker sock = mosq_test.do_client_connect( connect_packet, connack_packet2, timeout=5, port=bridge_target_port ) - mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback") # Restart bridging broker broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port) # Check, if all message got forwarded through the bridge for i in range(num_messages): - payload = f"queued message {i:3}" - mid = 1 + i - publish_packet = mosq_test.gen_publish( - topic, - mid=mid, - qos=qos, - payload=payload.encode("UTF-8"), - proto_ver=proto_ver, - ) - puback_packet = mosq_test.gen_puback(mid=mid, proto_ver=proto_ver) + publish_packet, puback_packet = gen_pub_packets(idx=i, mid_offset=1) mosq_test.do_receive_send( sock, publish_packet, @@ -189,11 +203,19 @@ def do_test(test_case_name: str, bridging_add_config: dict, target_add_config: d mosq_test.do_ping(sock) sock.close() + # Reconnect to the bridging broker and send ping to make sure the broker has process + # PUBACK from bridge target before getting shut down + sock = mosq_test.do_client_connect( + connect2_packet, connack2_packet, timeout=5, port=port + ) + mosq_test.do_ping(sock) + sock.close() + # Stop both brokers - (broker_terminate_rc, stde) = mosq_test.terminate_broker(broker) - broker = None (broker_terminate_rc, stde2) = mosq_test.terminate_broker(bridge_target_broker) bridge_target_broker = None + (broker_terminate_rc, stde) = mosq_test.terminate_broker(broker) + broker = None persist_help.check_counts( port, @@ -224,23 +246,32 @@ def do_test(test_case_name: str, bridging_add_config: dict, target_add_config: d print(f"{test_case_name}") if rc: - print(stde.decode("utf-8")) + if stde2 is not None: + print("Bridge target brocker log:") + print(stde2.decode("utf-8")) + if stde3 is not None: + print("Bridging brocker log (first run):") + print(stde3.decode("utf-8")) + if stde is not None: + print("Bridging brocker log:") + print(stde.decode("utf-8")) assert rc == 0, f"rc: {rc}" + in_bridge_config = { - "connection": "in-bridge", - "address": f"localhost:{port}", - "local_clientid": "bridge-test", - "remote_clientid": "upstream-bridge", - "topic": f"{topic} in 2", + "connection": "in-bridge", + "address": f"localhost:{port}", + "local_clientid": "bridge-test", + "remote_clientid": "upstream-bridge", + "topic": f"{topic} in 2", } out_bridge_config = { - "connection": "out-bridge", - "address": f"localhost:{bridge_target_port}", - "local_clientid": "upstream-bridge", - "remote_clientid": "bridge-test", - "topic": f"{topic} out 2", + "connection": "out-bridge", + "address": f"localhost:{bridge_target_port}", + "local_clientid": "upstream-bridge", + "remote_clientid": "bridge-test", + "topic": f"{topic} out 2", } memory_queue_config = { @@ -250,7 +281,7 @@ memory_queue_config = { do_test( "memory queue out bridge", - bridging_add_config= memory_queue_config | out_bridge_config, + bridging_add_config=memory_queue_config | out_bridge_config, target_add_config={}, ) diff --git a/test/broker/15-persist-client-drop-expired-messages.py b/test/broker/15-persist-client-drop-expired-messages.py index 48a37214..87d4970a 100755 --- a/test/broker/15-persist-client-drop-expired-messages.py +++ b/test/broker/15-persist-client-drop-expired-messages.py @@ -11,6 +11,7 @@ port = mosq_test.get_port() num_messages = 100 + def do_test(test_case_name: str, additional_config_entries: dict): conf_file = os.path.basename(__file__).replace(".py", f"_{port}.conf") persist_help.write_config( @@ -29,7 +30,11 @@ def do_test(test_case_name: str, additional_config_entries: dict): proto_ver = 5 connect_packet = mosq_test.gen_connect( - client_id, username=username, proto_ver=proto_ver, clean_session=False, session_expiry=60 + client_id, + username=username, + proto_ver=proto_ver, + clean_session=False, + session_expiry=60, ) connack_packet1 = mosq_test.gen_connack(rc=0, proto_ver=proto_ver) connack_packet2 = mosq_test.gen_connack(rc=0, flags=1, proto_ver=proto_ver) @@ -39,15 +44,13 @@ def do_test(test_case_name: str, additional_config_entries: dict): suback_packet = mosq_test.gen_suback(mid, qos=qos, proto_ver=proto_ver) connect2_packet = mosq_test.gen_connect( - source_id, username=username, proto_ver=proto_ver + source_id, username=username, proto_ver=proto_ver ) connack2_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver) rc = 1 - broker = mosq_test.start_broker( - filename=conf_file, use_conf=True, port=port - ) + broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port) con = None try: @@ -60,7 +63,9 @@ def do_test(test_case_name: str, additional_config_entries: dict): sock = mosq_test.do_client_connect( connect2_packet, connack2_packet, timeout=5, port=port ) - props = mqtt5_props.gen_uint32_prop(mqtt5_props.PROP_MESSAGE_EXPIRY_INTERVAL, 60) + props = mqtt5_props.gen_uint32_prop( + mqtt5_props.PROP_MESSAGE_EXPIRY_INTERVAL, 60 + ) for i in range(num_messages): payload = f"queued message {i:3}" mid = 10 + i @@ -140,20 +145,20 @@ def do_test(test_case_name: str, additional_config_entries: dict): qos, 0, persist_help.ms_queued, - idx=i, ) # Modify message expiry timestamp in the database assert persist_help.modify_base_msgs(port, sub_expiry_time=120) == num_messages # Restart broker - broker = mosq_test.start_broker( - filename=conf_file, use_conf=True, port=port - ) + broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port) # Connect client again, it should have a session, but all queued messages should be dropped sock = mosq_test.do_client_connect( - connect_packet, connack_packet2, timeout=5, port=port, + connect_packet, + connack_packet2, + timeout=5, + port=port, ) # Send ping and wait for the PINGRESP to make sure the broker will not send a queued message instead diff --git a/test/broker/15-persist-client-expired-session.py b/test/broker/15-persist-client-expired-session.py index d42d92b1..b57d1445 100755 --- a/test/broker/15-persist-client-expired-session.py +++ b/test/broker/15-persist-client-expired-session.py @@ -149,7 +149,6 @@ def do_test( qos, 0, persist_help.ms_queued, - idx=i, ) # Put session expiry_time into the past diff --git a/test/broker/15-persist-client-msg-modify-acl.py b/test/broker/15-persist-client-msg-modify-acl.py index 00722b69..76c34d36 100755 --- a/test/broker/15-persist-client-msg-modify-acl.py +++ b/test/broker/15-persist-client-msg-modify-acl.py @@ -53,9 +53,7 @@ def do_test(test_case_name: str, additional_config_entries: dict): rc = 1 - broker = mosq_test.start_broker( - filename=os.path.basename(__file__), use_conf=True, port=port - ) + broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port) con = None try: @@ -146,7 +144,6 @@ def do_test(test_case_name: str, additional_config_entries: dict): qos, 0, persist_help.ms_queued, - idx=i, ) # Remove any permission for the test topic and the test user @@ -156,9 +153,7 @@ def do_test(test_case_name: str, additional_config_entries: dict): os.chmod(f"{acl_file}", 0o644) # Restart broker - broker = mosq_test.start_broker( - filename=os.path.basename(__file__), use_conf=True, port=port - ) + broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port) # Connect client again, it should have a session, but all queued messages should be dropped sock = mosq_test.do_client_connect( diff --git a/test/broker/mosq_test_helper.py b/test/broker/mosq_test_helper.py index aa5c1e6e..2d2450a4 100644 --- a/test/broker/mosq_test_helper.py +++ b/test/broker/mosq_test_helper.py @@ -1,7 +1,11 @@ import inspect, os, sys # From http://stackoverflow.com/questions/279237/python-import-a-module-from-a-folder -cmd_subfolder = os.path.realpath(os.path.abspath(os.path.join(os.path.split(inspect.getfile( inspect.currentframe() ))[0],".."))) +cmd_subfolder = os.path.realpath( + os.path.abspath( + os.path.join(os.path.split(inspect.getfile(inspect.currentframe()))[0], "..") + ) +) if cmd_subfolder not in sys.path: sys.path.insert(0, cmd_subfolder) @@ -23,48 +27,67 @@ ssl_dir = source_dir.parent / "ssl" import importlib + def persist_module(): if len(sys.argv) > 1: - mod = sys.argv.pop(1) + mod = sys.argv.pop(1).replace(".py", "") else: raise RuntimeError("Not enough command line arguments - need persist module") return importlib.import_module(mod) -def do_test_broker_failure(conf_file : str, config : list, port : int, rc_expected : int, error_log_entry : str = None, stdout_entry : str =None, cmd_args : list = None): + +def do_test_broker_failure( + conf_file: str, + config: list, + port: int, + rc_expected: int, + error_log_entry: str = None, + stdout_entry: str = None, + cmd_args: list = None, +): rc = 1 use_conf_file = len(conf_file) create_conf_file = use_conf_file and len(config) if create_conf_file: - with open(conf_file, 'w') as f: + with open(conf_file, "w") as f: f.write("\n".join(config)) f.write("\n") try: broker = None - broker = mosq_test.start_broker(conf_file, port=port, use_conf=use_conf_file, expect_fail=True, cmd_args=cmd_args) + broker = mosq_test.start_broker( + conf_file, + port=port, + use_conf=use_conf_file, + expect_fail=True, + cmd_args=cmd_args, + ) (stdo, stde) = broker.communicate() if broker.returncode != rc_expected: print(f"Expected broker return code {rc_expected}, got {broker.returncode}") - (stdo, stde) = broker.communicate() - print(stde.decode('utf-8')) + print(stde.decode("utf-8")) return rc if error_log_entry is not None: - error_log = stde.decode('utf-8') + error_log = stde.decode("utf-8") if error_log_entry not in error_log: - print(f"Error log entry: '{error_log_entry}' not found in '{error_log}'") + print( + f"Error log entry: '{error_log_entry}' not found in '{error_log}'" + ) return rc if stdout_entry is not None: - stdout_log = stdo.decode('utf-8') + stdout_log = stdo.decode("utf-8") if stdout_entry not in stdout_log: - print(f"Error stdout entry: '{stdout_entry}' not found in '{stdout_log}'") + print( + f"Error stdout entry: '{stdout_entry}' not found in '{stdout_log}'" + ) return rc rc = 0 except subprocess.TimeoutExpired: if broker is not None: - mosq_test.wait_for_subprocess(broker,timeout=1) + mosq_test.wait_for_subprocess(broker, timeout=1) return rc except Exception as e: print(e) @@ -76,8 +99,9 @@ def do_test_broker_failure(conf_file : str, config : list, port : int, rc_expect except FileNotFoundError: pass if rc: - print(f"While testing 'config {chr(10).join(config) if len(config) else ''}'{', args'+ ' '.join(cmd_args) if cmd_args is not None else ''}") + print( + f"While testing 'config {chr(10).join(config) if len(config) else ''}'{', args'+ ' '.join(cmd_args) if cmd_args is not None else ''}" + ) exit(rc) - return rc; - + return rc diff --git a/test/broker/persist_sqlite.py b/test/broker/persist_sqlite.py index 9c5a7bc7..c42ea2ec 100755 --- a/test/broker/persist_sqlite.py +++ b/test/broker/persist_sqlite.py @@ -20,7 +20,8 @@ ms_send_pubrec = 10 ms_queued = 11 -def write_config(filename, port, additional_config_entries : dict = {}): + +def write_config(filename, port, additional_config_entries: dict = {}): with open(filename, "w") as f: f.write("listener %d\n" % (port)) f.write("allow_anonymous true\n") @@ -78,6 +79,7 @@ def init(port, create_db_of_version: list[int] = None): # We need to set write permission to everybody as broker will start with privilege drop os.chmod(f"{port}/mosquitto.sqlite3", 0o666) + def cleanup(port): rc = 1 try: @@ -118,6 +120,7 @@ def check_version_infos(port, database_schema_version): assert row[i] == database_schema_version[i] con.close() + def check_counts( port, clients=0, @@ -190,10 +193,14 @@ def check_client( "SELECT client_id, username, will_delay_time, session_expiry_time, " + "listener_port, max_packet_size, max_qos, retain_available, " + "session_expiry_interval, will_delay_interval " - + "FROM clients" + + "FROM clients " + + f"WHERE client_id = '{client_id}'" ) row = cur.fetchone() + if row is None: + raise ValueError(f"Cannot find client {client_id} in db") + if row[0] != client_id: raise ValueError("Invalid client_id %s / %s" % (row[0], client_id)) @@ -236,20 +243,17 @@ def check_client( ) con.close() -def modify_client( - port : int, - client_id : str, - sub_expiry_time : int -): + +def modify_client(port: int, client_id: str, sub_expiry_time: int): num_modified_rows = 0 con = sqlite3.connect(f"{port}/mosquitto.sqlite3") try: cur = con.cursor() cur.execute( "UPDATE clients" - +f" SET session_expiry_time = session_expiry_time - {sub_expiry_time}" - +f" WHERE client_id = ?", - (client_id,) + + f" SET session_expiry_time = session_expiry_time - {sub_expiry_time}" + + f" WHERE client_id = ?", + (client_id,), ) num_modified_rows = cur.rowcount con.commit() @@ -258,6 +262,7 @@ def modify_client( return num_modified_rows + def check_subscription( port, client_id, topic, subscription_options, subscription_identifier ): @@ -265,10 +270,14 @@ def check_subscription( cur = con.cursor() cur.execute( "SELECT client_id, topic, subscription_options, subscription_identifier " - + "FROM subscriptions" + + "FROM subscriptions " + + f"WHERE client_id = '{client_id}'" ) row = cur.fetchone() + if row is None: + raise ValueError(f"Cannot find client {client_id} in db") + if row[0] != client_id: raise ValueError("Invalid client_id %s / %s" % (row[0], client_id)) @@ -289,7 +298,7 @@ def check_subscription( def check_client_msg( - port, client_id, cmsg_id, store_id, dup, direction, mid, qos, retain, state, idx=0 + port, client_id, cmsg_id, store_id, dup, direction, mid, qos, retain, state ): con = sqlite3.connect(f"{port}/mosquitto.sqlite3") try: @@ -297,39 +306,64 @@ def check_client_msg( cur.execute( "SELECT client_id,cmsg_id,store_id,dup,direction,mid,qos,retain,state " + "FROM client_msgs " - + "ORDER BY cmsg_id" + + f"WHERE client_id = '{client_id}' AND cmsg_id = {cmsg_id}" ) - for i in range(0, idx + 1): - row = cur.fetchone() + row = cur.fetchone() + + msg_id = f"client_id={client_id},cmsg_id={cmsg_id}" + if row is None: + raise ValueError( + f"Cannot find client message client_id = {client_id} cmsg_id = {msg_id} in db." + ) if row[0] != client_id: - raise ValueError("Invalid client_id %s / %s" % (row[0], client_id)) + raise ValueError( + "Invalid client_id %s / %s for message %s" % (row[0], client_id, msg_id) + ) if row[1] != cmsg_id: - raise ValueError("Invalid cmsg_id %s / %s" % (row[1], cmsg_id)) + raise ValueError( + "Invalid cmsg_id %s / %s for message %s" % (row[1], cmsg_id, msg_id) + ) if row[2] != store_id: - raise ValueError("Invalid store_id %d / %d" % (row[2], store_id)) + raise ValueError( + "Invalid store_id %d / %d for message %s" % (row[2], store_id, msg_id) + ) if row[3] != dup: - raise ValueError("Invalid dup %d / %d" % (row[3], dup)) + raise ValueError( + "Invalid dup %d / %d for message %s" % (row[3], dup, msg_id) + ) if row[4] != direction: - raise ValueError("Invalid direction %d / %d" % (row[4], direction)) + raise ValueError( + "Invalid direction %d / %d for message %s" % (row[4], direction, msg_id) + ) if row[5] != mid: - raise ValueError("Invalid mid %d / %d" % (row[5], mid)) + raise ValueError( + "Invalid mid %d / %d for message %s" % (row[5], mid, msg_id) + ) if row[6] != qos: - raise ValueError("Invalid qos %d / %d" % (row[6], qos)) + raise ValueError( + "Invalid qos %d / %d for message %s" % (row[6], qos, msg_id) + ) if row[7] != retain: - raise ValueError("Invalid retain %d / %d" % (row[7], retain)) + raise ValueError( + "Invalid retain %d / %d for message %s" % (row[7], retain, msg_id) + ) if row[8] != state: - raise ValueError("Invalid state %d / %d" % (row[8], state)) + raise ValueError( + "Invalid state %d / %d for message %s" % (row[8], state, msg_id) + ) except ValueError as err: - raise ValueError(str(err)+ f" at index {idx}") from err + raise ValueError( + str(err) + f" in client message client_id = {client_id} cmsg_id = {idx}" + ) from err finally: con.close() @@ -359,6 +393,9 @@ def check_base_msg( for i in range(0, idx + 1): row = cur.fetchone() + if row is None: + raise ValueError(f"no base messages") + if row[0] == 0: raise ValueError("Invalid store_id %d / %d" % (row[0], store_id)) @@ -375,7 +412,9 @@ def check_base_msg( raise ValueError("Invalid source_id %s / %s" % (row[4], source_id)) if row[5] != source_username: - raise ValueError("Invalid source_username %s / %s" % (row[5], source_username)) + raise ValueError( + "Invalid source_username %s / %s" % (row[5], source_username) + ) if row[6] != payloadlen or (payloadlen != 0 and row[6] != len(row[3])): raise ValueError("Invalid payloadlen %d / %d" % (row[6], payloadlen)) @@ -392,23 +431,23 @@ def check_base_msg( if row[10] != retain: raise ValueError("Invalid retain %d / %d" % (row[10], retain)) except ValueError as err: - raise ValueError(str(err)+ f" at index {idx}") from err + raise ValueError(str(err) + f" at index {idx}") from err finally: con.close() return row[0] + def modify_base_msgs( - port : int, - sub_expiry_time : int + port: int, + sub_expiry_time: int, ): num_modified_rows = 0 con = sqlite3.connect(f"{port}/mosquitto.sqlite3") try: cur = con.cursor() cur.execute( - "UPDATE base_msgs" - +f" SET expiry_time = expiry_time - {sub_expiry_time}" + "UPDATE base_msgs" + f" SET expiry_time = expiry_time - {sub_expiry_time}" ) num_modified_rows = cur.rowcount con.commit() @@ -416,6 +455,7 @@ def modify_base_msgs( con.close() return num_modified_rows + def check_retain(port, topic, store_id): con = sqlite3.connect(f"{port}/mosquitto.sqlite3") cur = con.cursor() From 6ca0fe1a147abff43067131561da533bf2b3932b Mon Sep 17 00:00:00 2001 From: Norbert Heusser Date: Mon, 11 Mar 2024 14:30:00 +0000 Subject: [PATCH 5/6] Add second subscriber to persistent plugin expiry test Signed-off-by: Norbert Heusser --- ...15-persist-client-drop-expired-messages.py | 234 ++++++++-------- .../15-persist-client-expired-session.py | 262 +++++++++--------- test/broker/persist_module_helper.py | 157 +++++++++++ 3 files changed, 400 insertions(+), 253 deletions(-) create mode 100644 test/broker/persist_module_helper.py diff --git a/test/broker/15-persist-client-drop-expired-messages.py b/test/broker/15-persist-client-drop-expired-messages.py index 87d4970a..84b0d671 100755 --- a/test/broker/15-persist-client-drop-expired-messages.py +++ b/test/broker/15-persist-client-drop-expired-messages.py @@ -4,6 +4,7 @@ # different client, restore, reconnect, check it is received. from mosq_test_helper import * +from persist_module_helper import * persist_help = persist_module() @@ -11,8 +12,26 @@ port = mosq_test.get_port() num_messages = 100 +proto_ver = 5 +qos = 1 +topic = "test-expired-msgs" +username = "test-message-expiry" + +subscriber_id = "test-subscriber" +second_subscriber_id = "second-subscriber" +publisher_id = "test-publisher" + + +def do_test( + test_case_name: str, + additional_config_entries: dict, + resubscribe: bool, + num_messages_two_subscribers: int = 0, +): + print( + f"{test_case_name}, resubscribe = {resubscribe}, two_subscribers = {'True' if num_messages_two_subscribers > 0 else 'False'}" + ) -def do_test(test_case_name: str, additional_config_entries: dict): conf_file = os.path.basename(__file__).replace(".py", f"_{port}.conf") persist_help.write_config( conf_file, @@ -21,32 +40,9 @@ def do_test(test_case_name: str, additional_config_entries: dict): ) persist_help.init(port) - client_id = "test-expired-messages-subscriber" - username = "test-message-expiry" - - qos = 1 - topic = "client-msg/test" - source_id = "test-expired-messages-publisher" - proto_ver = 5 - - connect_packet = mosq_test.gen_connect( - client_id, - username=username, - proto_ver=proto_ver, - clean_session=False, - session_expiry=60, - ) - connack_packet1 = mosq_test.gen_connack(rc=0, proto_ver=proto_ver) - connack_packet2 = mosq_test.gen_connack(rc=0, flags=1, proto_ver=proto_ver) - - mid = 1 - subscribe_packet = mosq_test.gen_subscribe(mid, topic, qos, proto_ver=proto_ver) - suback_packet = mosq_test.gen_suback(mid, qos=qos, proto_ver=proto_ver) - connect2_packet = mosq_test.gen_connect( - source_id, username=username, proto_ver=proto_ver + publisher_id, username=username, proto_ver=proto_ver ) - connack2_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver) rc = 1 @@ -54,126 +50,99 @@ def do_test(test_case_name: str, additional_config_entries: dict): con = None try: - sock = mosq_test.do_client_connect( - connect_packet, connack_packet1, timeout=5, port=port - ) - mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback") - sock.close() + msg_counts = {subscriber_id: num_messages} - sock = mosq_test.do_client_connect( - connect2_packet, connack2_packet, timeout=5, port=port + connect_client( + port, + subscriber_id, + username, + proto_ver, + session_expiry=60, + subscribe_topic=topic, + ).close() + + publisher_sock = connect_client( + port, publisher_id, username, proto_ver, session_expiry=0 ) - props = mqtt5_props.gen_uint32_prop( - mqtt5_props.PROP_MESSAGE_EXPIRY_INTERVAL, 60 + publish_messages( + publisher_sock, + proto_ver, + topic, + 0, + num_messages - num_messages_two_subscribers, + message_expiry=60, ) - for i in range(num_messages): - payload = f"queued message {i:3}" - mid = 10 + i - publish_packet = mosq_test.gen_publish( + + if num_messages_two_subscribers > 0: + msg_counts[second_subscriber_id] = num_messages_two_subscribers + connect_client( + port, + second_subscriber_id, + username, + proto_ver, + session_expiry=60, + subscribe_topic=topic, + ).close() + publish_messages( + publisher_sock, + proto_ver, topic, - mid=mid, - qos=qos, - payload=payload.encode("UTF-8"), - proto_ver=proto_ver, - properties=props, + num_messages - num_messages_two_subscribers, + num_messages, + message_expiry=60, ) - puback_packet = mosq_test.gen_puback(mid=mid, proto_ver=proto_ver) - mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback") - sock.close() + publisher_sock.close() # Terminate the broker (broker_terminate_rc, stde) = mosq_test.terminate_broker(broker) broker = None - persist_help.check_counts( + check_db( + persist_help, port, - clients=1, - client_msgs_out=num_messages, - base_msgs=num_messages, - subscriptions=1, + username, + subscription_topic=topic, + client_msg_counts=msg_counts, + publisher_id=publisher_id, + num_published_msgs=num_messages, + message_expiry=60, ) - # Check client - # persist_help.check_client( - # port, - # client_id, - # username=username, - # will_delay_time=0, - # session_expiry_time=60, - # listener_port=port, - # max_packet_size=0, - # max_qos=2, - # retain_available=1, - # session_expiry_interval=60, - # will_delay_interval=0, - # ) - - # Check subscription - persist_help.check_subscription(port, client_id, topic, qos, 0) - - # Check stored message - for i in range(num_messages): - payload = f"queued message {i:3}" - payload_b = payload.encode("UTF-8") - mid = 10 + i - store_id = persist_help.check_base_msg( - port, - 60, - topic, - payload_b, - source_id, - username, - len(payload_b), - mid, - port, - qos, - retain=0, - idx=i, - ) - - # Check client msg - subscriber_mid = 1 + i - cmsg_id = 1 + i - persist_help.check_client_msg( - port, - client_id, - cmsg_id, - store_id, - 0, - persist_help.dir_out, - subscriber_mid, - qos, - 0, - persist_help.ms_queued, - ) - - # Modify message expiry timestamp in the database + # Put session expiry_time into the past assert persist_help.modify_base_msgs(port, sub_expiry_time=120) == num_messages # Restart broker broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port) - # Connect client again, it should have a session, but all queued messages should be dropped - sock = mosq_test.do_client_connect( - connect_packet, - connack_packet2, - timeout=5, - port=port, - ) - + # Reconnect client(s), it should have a session, but all queued messages should be dropped + for client_id in msg_counts.keys(): + subscriber_sock = connect_client( + port, + client_id, + username, + proto_ver, + session_expiry=60, + session_present=True, + subscribe_topic=topic if resubscribe else None, + ) # Send ping and wait for the PINGRESP to make sure the broker will not send a queued message instead - mosq_test.do_ping(sock) - sock.close() + mosq_test.do_ping(subscriber_sock) + subscriber_sock.close() (broker_terminate_rc, stde) = mosq_test.terminate_broker(broker) broker = None - persist_help.check_counts( + for client_id in msg_counts.keys(): + # None for subscriber with subscriber_id means no subscription + msg_counts[client_id] = 0 + check_db( + persist_help, port, - clients=1, - client_msgs_out=0, - base_msgs=0, - subscriptions=1, + username, + subscription_topic=topic, + client_msg_counts=msg_counts, + publisher_id=publisher_id, + num_published_msgs=num_messages, ) rc = broker_terminate_rc @@ -187,16 +156,29 @@ def do_test(test_case_name: str, additional_config_entries: dict): os.remove(conf_file) rc += persist_help.cleanup(port) - print(f"{test_case_name}") if rc: print(stde.decode("utf-8")) assert rc == 0, f"rc: {rc}" +memory_queue_config = { + "log_type": "all", + "max_queued_messages": num_messages, +} + do_test( "memory queue", - additional_config_entries={ - "log_type": "all", - "max_queued_messages": num_messages, - }, + additional_config_entries=memory_queue_config, + resubscribe=False, +) +do_test( + "memory queue", + additional_config_entries=memory_queue_config, + resubscribe=True, +) +do_test( + "memory queue", + additional_config_entries=memory_queue_config, + resubscribe=False, + num_messages_two_subscribers=30, ) diff --git a/test/broker/15-persist-client-expired-session.py b/test/broker/15-persist-client-expired-session.py index b57d1445..a28a6574 100755 --- a/test/broker/15-persist-client-expired-session.py +++ b/test/broker/15-persist-client-expired-session.py @@ -4,17 +4,33 @@ # different client, restore, reconnect, check it is received. from mosq_test_helper import * +from persist_module_helper import * persist_help = persist_module() port = mosq_test.get_port() num_messages = 100 +proto_ver = 5 +qos = 1 +topic = "test-session-expiry" +username = "test-session-expiry" + +subscriber_id = "test-expired-session-subscriber" +second_subscriber_id = "second-subscriber" +publisher_id = "test-expired-session-publisher" def do_test( - test_case_name: str, additional_config_entries: dict, message_expiry_interval: int + test_case_name: str, + additional_config_entries: dict, + resubscribe: bool, + num_messages_two_subscribers: int = 0, ): + print( + f"{test_case_name}, resubscribe = {resubscribe}, two_subscribers = {'True' if num_messages_two_subscribers > 0 else 'False'}" + ) + conf_file = os.path.basename(__file__).replace(".py", f"_{port}.conf") persist_help.write_config( conf_file, @@ -23,29 +39,8 @@ def do_test( ) persist_help.init(port) - client_id = "test-expired-session-subscriber" - username = "test-session-expiry" - - qos = 1 - topic = "client-msg/test" - source_id = "test-expired-session-publisher" - proto_ver = 5 - - connect_packet = mosq_test.gen_connect( - client_id, - username=username, - proto_ver=proto_ver, - clean_session=False, - session_expiry=60, - ) - connack_packet = mosq_test.gen_connack(rc=0, proto_ver=proto_ver) - - mid = 1 - subscribe_packet = mosq_test.gen_subscribe(mid, topic, qos, proto_ver=proto_ver) - suback_packet = mosq_test.gen_suback(mid, qos=qos, proto_ver=proto_ver) - connect2_packet = mosq_test.gen_connect( - source_id, username=username, proto_ver=proto_ver + publisher_id, username=username, proto_ver=proto_ver ) rc = 1 @@ -54,132 +49,134 @@ def do_test( con = None try: - sock = mosq_test.do_client_connect( - connect_packet, connack_packet, timeout=5, port=port - ) - mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback") - sock.close() + msg_counts = {subscriber_id: num_messages} - sock = mosq_test.do_client_connect( - connect2_packet, connack_packet, timeout=5, port=port + connect_client( + port, + subscriber_id, + username, + proto_ver, + session_expiry=60, + subscribe_topic=topic, + ).close() + + publisher_sock = connect_client( + port, publisher_id, username, proto_ver, session_expiry=0 ) - props = ( - mqtt5_props.gen_uint32_prop( - mqtt5_props.PROP_MESSAGE_EXPIRY_INTERVAL, message_expiry_interval - ) - if message_expiry_interval > 0 - else b"" + publish_messages( + publisher_sock, + proto_ver, + topic, + 0, + num_messages - num_messages_two_subscribers, ) - for i in range(num_messages): - payload = f"queued message {i:3}" - mid = 10 + i - publish_packet = mosq_test.gen_publish( + + if num_messages_two_subscribers > 0: + msg_counts[second_subscriber_id] = num_messages_two_subscribers + connect_client( + port, + second_subscriber_id, + username, + proto_ver, + session_expiry=60, + subscribe_topic=topic, + ).close() + publish_messages( + publisher_sock, + proto_ver, topic, - mid=mid, - qos=qos, - payload=payload.encode("UTF-8"), - proto_ver=proto_ver, - properties=props, + num_messages - num_messages_two_subscribers, + num_messages, ) - puback_packet = mosq_test.gen_puback(mid=mid, proto_ver=proto_ver) - mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback") - sock.close() + publisher_sock.close() # Terminate the broker (broker_terminate_rc, stde) = mosq_test.terminate_broker(broker) broker = None - persist_help.check_counts( + check_db( + persist_help, port, - clients=1, - client_msgs_out=num_messages, - base_msgs=num_messages, - subscriptions=1, + username, + subscription_topic=topic, + client_msg_counts=msg_counts, + publisher_id=publisher_id, + num_published_msgs=num_messages, ) - # Check client - persist_help.check_client( - port, - client_id, - username=username, - will_delay_time=0, - session_expiry_time=60, - listener_port=port, - max_packet_size=0, - max_qos=2, - retain_available=1, - session_expiry_interval=60, - will_delay_interval=0, - ) - - # Check subscription - persist_help.check_subscription(port, client_id, topic, qos, 0) - - # Check stored message - for i in range(num_messages): - payload = f"queued message {i:3}" - payload_b = payload.encode("UTF-8") - mid = 10 + i - store_id = persist_help.check_base_msg( - port, - message_expiry_interval, - topic, - payload_b, - source_id, - username, - len(payload_b), - mid, - port, - qos, - retain=0, - idx=i, - ) - - # Check client msg - subscriber_mid = 1 + i - cmsg_id = 1 + i - persist_help.check_client_msg( - port, - client_id, - cmsg_id, - store_id, - 0, - persist_help.dir_out, - subscriber_mid, - qos, - 0, - persist_help.ms_queued, - ) - # Put session expiry_time into the past - assert persist_help.modify_client(port, client_id, sub_expiry_time=120) == 1 + assert persist_help.modify_client(port, subscriber_id, sub_expiry_time=120) == 1 # Restart broker broker = mosq_test.start_broker(filename=conf_file, use_conf=True, port=port) - # Connect client again, it should have a session, but all queued messages should be dropped - sock = mosq_test.do_client_connect( - connect_packet, - connack_packet, - timeout=5, - port=port, + # Reconnect client, it should have a session, but all queued messages should be dropped + subscriber_sock = connect_client( + port, + subscriber_id, + username, + proto_ver, + session_expiry=60, + subscribe_topic=topic if resubscribe else None, ) - # Send ping and wait for the PINGRESP to make sure the broker will not send a queued message instead - mosq_test.do_ping(sock) - sock.close() + mosq_test.do_ping(subscriber_sock) + subscriber_sock.close() (broker_terminate_rc, stde) = mosq_test.terminate_broker(broker) broker = None - persist_help.check_counts( + # None for subscriber with subscriber_id means no subscription + msg_counts[subscriber_id] = 0 if resubscribe else None + check_db( + persist_help, port, - clients=1, - client_msgs_out=0, - base_msgs=0, - subscriptions=0, + username, + subscription_topic=topic, + client_msg_counts=msg_counts, + publisher_id=publisher_id, + num_published_msgs=num_messages, ) + if num_messages_two_subscribers > 0: + # Put session expiry_time into the past + assert ( + persist_help.modify_client( + port, second_subscriber_id, sub_expiry_time=120 + ) + == 1 + ) + # Restart broker + broker = mosq_test.start_broker( + filename=conf_file, use_conf=True, port=port + ) + # Reconnect client, it should have a session, but all queued messages should be dropped + subscriber_sock = connect_client( + port, + second_subscriber_id, + username, + proto_ver, + session_expiry=60, + subscribe_topic=topic if resubscribe else None, + ) + # Send ping and wait for the PINGRESP to make sure the broker will not send a queued message instead + mosq_test.do_ping(subscriber_sock) + subscriber_sock.close() + + (broker_terminate_rc, stde) = mosq_test.terminate_broker(broker) + broker = None + + msg_counts[second_subscriber_id] = 0 if resubscribe else None + check_db( + persist_help, + port, + username, + subscription_topic=topic, + client_msg_counts=msg_counts, + publisher_id=publisher_id, + num_published_msgs=num_messages, + ) + rc = broker_terminate_rc finally: if broker is not None: @@ -191,7 +188,6 @@ def do_test( os.remove(conf_file) rc += persist_help.cleanup(port) - print(f"{test_case_name}") if rc: print(stde.decode("utf-8")) assert rc == 0, f"rc: {rc}" @@ -204,12 +200,24 @@ memory_queue_config = { do_test( - "memory queue, message expiry interval: 0", + "memory queue", additional_config_entries=memory_queue_config, - message_expiry_interval=0, + resubscribe=False, ) do_test( - "memory queue, message expiry interval: 120", + "memory queue", additional_config_entries=memory_queue_config, - message_expiry_interval=120, + resubscribe=True, +) +do_test( + "memory queue", + additional_config_entries=memory_queue_config, + resubscribe=False, + num_messages_two_subscribers=20, +) +do_test( + "memory queue", + additional_config_entries=memory_queue_config, + resubscribe=True, + num_messages_two_subscribers=20, ) diff --git a/test/broker/persist_module_helper.py b/test/broker/persist_module_helper.py new file mode 100644 index 00000000..15b9594a --- /dev/null +++ b/test/broker/persist_module_helper.py @@ -0,0 +1,157 @@ +#!/usr/bin/env python3 + +import socket +import mosq_test +import mqtt5_props + +from typing import Optional +from types import ModuleType + + +def connect_client( + port: int, + client_id: str, + username: str, + proto_ver: int, + session_expiry: int, + session_present: bool = False, + subscribe_topic: Optional[str] = None, + qos: int = 1, +): + connect_packet = mosq_test.gen_connect( + client_id=client_id, + username=username, + proto_ver=proto_ver, + clean_session=False, + session_expiry=session_expiry, + ) + connack_packet = mosq_test.gen_connack( + rc=0, proto_ver=proto_ver, flags=1 if session_present else 0 + ) + sock = mosq_test.do_client_connect( + connect_packet, connack_packet, timeout=5, port=port + ) + if subscribe_topic is not None: + mid = 1 + subscribe_packet = mosq_test.gen_subscribe( + mid, subscribe_topic, qos, proto_ver=proto_ver + ) + suback_packet = mosq_test.gen_suback(mid, qos=qos, proto_ver=proto_ver) + mosq_test.do_send_receive(sock, subscribe_packet, suback_packet, "suback") + return sock + + +def publish_messages( + sock: socket, + proto_ver: int, + topic: str, + start: int, + end: int, + message_expiry: int = 0, + qos: int = 1, +): + for i in range(start, end): + payload = f"queued message {i:3}" + mid = 10 + i + props = ( + mqtt5_props.gen_uint32_prop( + mqtt5_props.PROP_MESSAGE_EXPIRY_INTERVAL, message_expiry + ) + if message_expiry > 0 + else b"" + ) + publish_packet = mosq_test.gen_publish( + topic, + mid=mid, + qos=qos, + payload=payload.encode("UTF-8"), + proto_ver=proto_ver, + properties=props, + ) + puback_packet = mosq_test.gen_puback(mid=mid, proto_ver=proto_ver) + mosq_test.do_send_receive(sock, publish_packet, puback_packet, "puback") + + +def check_db( + persist_help: ModuleType, + port: int, + username: str, + subscription_topic: str, + client_msg_counts: dict[str, int], + publisher_id: str, + num_published_msgs: int, + message_expiry: int = 0, + qos: int = 1, +): + count_list = [v for v in client_msg_counts.values() if v] + [0] + num_base_msgs = max(count_list) + num_subscriptions = sum(1 for c in client_msg_counts.values() if c is not None) + num_client_msgs_out = sum(count_list) + persist_help.check_counts( + port, + clients=len(client_msg_counts), + client_msgs_out=num_client_msgs_out, + base_msgs=num_base_msgs, + subscriptions=num_subscriptions, + ) + + # Check client + for client_id, num_messages_for_client in client_msg_counts.items(): + persist_help.check_client( + port, + client_id, + username=username, + will_delay_time=0, + session_expiry_time=60, + listener_port=None, # persist-lmdb reset listener port to 0 on disconnect + max_packet_size=0, + max_qos=2, + retain_available=1, + session_expiry_interval=60, + will_delay_interval=0, + ) + # Check subscription + if num_messages_for_client is not None: + persist_help.check_subscription(port, client_id, subscription_topic, qos, 0) + + # Check stored message + for i in range(num_base_msgs): + msg_id = num_published_msgs - num_base_msgs + i + payload = f"queued message {msg_id:3}" + payload_b = payload.encode("UTF-8") + mid = 10 + msg_id + store_id = persist_help.check_base_msg( + port, + message_expiry, + subscription_topic, + payload_b, + publisher_id, + username, + len(payload_b), + mid, + port, + qos, + retain=0, + idx=i, + ) + # Check client msg + for client_id, num_messages_for_client in client_msg_counts.items(): + if num_messages_for_client is None: + continue + client_msg_start = num_published_msgs - num_messages_for_client + if msg_id < client_msg_start: + continue + cmsg_id = 1 + msg_id - client_msg_start + subscriber_mid = cmsg_id + persist_help.check_client_msg( + port, + client_id, + cmsg_id, + store_id, + 0, + persist_help.dir_out, + subscriber_mid, + qos, + 0, + persist_help.ms_queued, + ) From fa6bb8b4586f7d88fe9bedec200aec6252f93161 Mon Sep 17 00:00:00 2001 From: Norbert Heusser Date: Mon, 11 Mar 2024 19:29:13 +0000 Subject: [PATCH 6/6] Fixed deletion of expired retain msg after restore from persistent plugin Signed-off-by: Norbert Heusser --- src/database.c | 6 ++++ src/mosquitto.c | 1 + src/mosquitto_broker_internal.h | 3 +- src/plugin_public.c | 1 - src/retain.c | 33 +++++++++++++++---- ...15-persist-client-drop-expired-messages.py | 13 +++++++- .../15-persist-client-expired-session.py | 32 +++++++++++++++++- test/broker/persist_module_helper.py | 10 ++++-- test/broker/persist_sqlite.py | 4 --- test/unit/broker/subs_stubs.c | 5 +++ 10 files changed, 90 insertions(+), 18 deletions(-) diff --git a/src/database.c b/src/database.c index 89f386e1..28b7aa82 100644 --- a/src/database.c +++ b/src/database.c @@ -1201,6 +1201,12 @@ int db__message_release_incoming(struct mosquitto *context, uint16_t mid) } +void db__retain_expiry_check() +{ + retain__expiry_check(&db.retains); +} + + void db__expire_all_messages(struct mosquitto *context) { struct mosquitto__client_msg *client_msg, *tmp; diff --git a/src/mosquitto.c b/src/mosquitto.c index 210e72ef..4a3789bc 100644 --- a/src/mosquitto.c +++ b/src/mosquitto.c @@ -427,6 +427,7 @@ int main(int argc, char *argv[]) plugin_persist__handle_restore(); session_expiry__check(); + db__retain_expiry_check(); db__msg_store_compact(); /* After loading persisted clients and ACLs, try to associate them, diff --git a/src/mosquitto_broker_internal.h b/src/mosquitto_broker_internal.h index 64d17dfa..5078236b 100644 --- a/src/mosquitto_broker_internal.h +++ b/src/mosquitto_broker_internal.h @@ -732,6 +732,7 @@ int db__message_write_queued_in(struct mosquitto *context); void db__msg_add_to_inflight_stats(struct mosquitto_msg_data *msg_data, struct mosquitto__client_msg *msg); void db__msg_add_to_queued_stats(struct mosquitto_msg_data *msg_data, struct mosquitto__client_msg *msg); uint64_t db__new_msg_id(void); +void db__retain_expiry_check(void); void db__expire_all_messages(struct mosquitto *context); void db__check_acl_of_all_messages(struct mosquitto *context); @@ -889,7 +890,7 @@ int retain__init(void); void retain__clean(struct mosquitto__retainhier **retainhier); int retain__queue(struct mosquitto *context, const struct mosquitto_subscription *sub); int retain__store(const char *topic, struct mosquitto__base_msg *base_msg, char **split_topics, bool persist); - +void retain__expiry_check(struct mosquitto__retainhier **retainhier); /* ============================================================ * Security related functions * ============================================================ */ diff --git a/src/plugin_public.c b/src/plugin_public.c index 3d9da0db..9531f074 100644 --- a/src/plugin_public.c +++ b/src/plugin_public.c @@ -751,7 +751,6 @@ BROKER_EXPORT int mosquitto_persist_base_msg_add(struct mosquitto_base_msg *msg_ { struct mosquitto context; struct mosquitto__base_msg *base_msg; - int i; int rc; memset(&context, 0, sizeof(context)); diff --git a/src/retain.c b/src/retain.c index 66bb37d6..e3718f20 100644 --- a/src/retain.c +++ b/src/retain.c @@ -192,6 +192,19 @@ int retain__store(const char *topic, struct mosquitto__base_msg *base_msg, char return MOSQ_ERR_SUCCESS; } +static bool retain__delete_expired_msg(struct mosquitto__retainhier *branch) +{ + if(branch->retained && branch->retained->data.expiry_time > 0 && db.now_real_s >= branch->retained->data.expiry_time){ + plugin_persist__handle_retain_msg_delete(branch->retained); + db__msg_store_ref_dec(&branch->retained); + branch->retained = NULL; +#ifdef WITH_SYS_TREE + db.retained_count--; +#endif + return true; + } + return false; +} static int retain__process(struct mosquitto__retainhier *branch, struct mosquitto *context, const struct mosquitto_subscription *sub) { @@ -200,13 +213,7 @@ static int retain__process(struct mosquitto__retainhier *branch, struct mosquitt uint16_t mid; struct mosquitto__base_msg *retained; - if(branch->retained->data.expiry_time > 0 && db.now_real_s >= branch->retained->data.expiry_time){ - plugin_persist__handle_retain_msg_delete(branch->retained); - db__msg_store_ref_dec(&branch->retained); - branch->retained = NULL; -#ifdef WITH_SYS_TREE - db.retained_count--; -#endif + if(retain__delete_expired_msg(branch)){ return MOSQ_ERR_SUCCESS; } @@ -344,6 +351,17 @@ int retain__queue(struct mosquitto *context, const struct mosquitto_subscription return MOSQ_ERR_SUCCESS; } +void retain__expiry_check(struct mosquitto__retainhier **retainhier) +{ + struct mosquitto__retainhier *peer, *retainhier_tmp; + + HASH_ITER(hh, *retainhier, peer, retainhier_tmp){ + retain__expiry_check(&peer->children); + if (retain__delete_expired_msg(peer)){ + retain__clean_empty_hierarchy(peer); + } + } +} void retain__clean(struct mosquitto__retainhier **retainhier) { @@ -360,3 +378,4 @@ void retain__clean(struct mosquitto__retainhier **retainhier) } } + diff --git a/test/broker/15-persist-client-drop-expired-messages.py b/test/broker/15-persist-client-drop-expired-messages.py index 84b0d671..7e5231e5 100755 --- a/test/broker/15-persist-client-drop-expired-messages.py +++ b/test/broker/15-persist-client-drop-expired-messages.py @@ -27,9 +27,10 @@ def do_test( additional_config_entries: dict, resubscribe: bool, num_messages_two_subscribers: int = 0, + num_retain_messages : int = 0, ): print( - f"{test_case_name}, resubscribe = {resubscribe}, two_subscribers = {'True' if num_messages_two_subscribers > 0 else 'False'}" + f"{test_case_name}, resubscribe = {resubscribe}, two_subscribers = {'True' if num_messages_two_subscribers > 0 else 'False'}, num_retain_messages = {num_retain_messages} " ) conf_file = os.path.basename(__file__).replace(".py", f"_{port}.conf") @@ -71,6 +72,7 @@ def do_test( 0, num_messages - num_messages_two_subscribers, message_expiry=60, + retain_end=num_retain_messages, ) if num_messages_two_subscribers > 0: @@ -90,6 +92,7 @@ def do_test( num_messages - num_messages_two_subscribers, num_messages, message_expiry=60, + retain_end=num_retain_messages, ) publisher_sock.close() @@ -105,6 +108,7 @@ def do_test( client_msg_counts=msg_counts, publisher_id=publisher_id, num_published_msgs=num_messages, + retain_end = num_retain_messages, message_expiry=60, ) @@ -143,6 +147,7 @@ def do_test( client_msg_counts=msg_counts, publisher_id=publisher_id, num_published_msgs=num_messages, + retain_end = 0, ) rc = broker_terminate_rc @@ -182,3 +187,9 @@ do_test( resubscribe=False, num_messages_two_subscribers=30, ) +do_test( + "memory queue", + additional_config_entries=memory_queue_config, + resubscribe=False, + num_retain_messages=40, +) diff --git a/test/broker/15-persist-client-expired-session.py b/test/broker/15-persist-client-expired-session.py index a28a6574..c8826e09 100755 --- a/test/broker/15-persist-client-expired-session.py +++ b/test/broker/15-persist-client-expired-session.py @@ -26,9 +26,10 @@ def do_test( additional_config_entries: dict, resubscribe: bool, num_messages_two_subscribers: int = 0, + num_retain_messages : int = 0, ): print( - f"{test_case_name}, resubscribe = {resubscribe}, two_subscribers = {'True' if num_messages_two_subscribers > 0 else 'False'}" + f"{test_case_name}, resubscribe = {resubscribe}, two_subscribers = {'True' if num_messages_two_subscribers > 0 else 'False'}, num_retain_messages = {num_retain_messages} " ) conf_file = os.path.basename(__file__).replace(".py", f"_{port}.conf") @@ -69,6 +70,7 @@ def do_test( topic, 0, num_messages - num_messages_two_subscribers, + retain_end=num_retain_messages, ) if num_messages_two_subscribers > 0: @@ -87,6 +89,7 @@ def do_test( topic, num_messages - num_messages_two_subscribers, num_messages, + retain_end=num_retain_messages, ) publisher_sock.close() @@ -102,6 +105,7 @@ def do_test( client_msg_counts=msg_counts, publisher_id=publisher_id, num_published_msgs=num_messages, + retain_end = num_retain_messages, ) # Put session expiry_time into the past @@ -136,6 +140,7 @@ def do_test( client_msg_counts=msg_counts, publisher_id=publisher_id, num_published_msgs=num_messages, + retain_end=num_retain_messages, ) if num_messages_two_subscribers > 0: @@ -175,6 +180,7 @@ def do_test( client_msg_counts=msg_counts, publisher_id=publisher_id, num_published_msgs=num_messages, + retain_end=num_retain_messages, ) rc = broker_terminate_rc @@ -221,3 +227,27 @@ do_test( resubscribe=True, num_messages_two_subscribers=20, ) +do_test( + "memory queue", + additional_config_entries=memory_queue_config, + resubscribe=False, + num_retain_messages=30, +) +# The following test case is open for discussion as adapting +# the check routines will be hard and some observations +# are unclear right now +# do_test( +# "memory queue", +# additional_config_entries=memory_queue_config, +# resubscribe=False, +# num_messages_two_subscribers=40, +# num_retain_messages=30, +# ) +# do_test( +# "memory queue", +# additional_config_entries=memory_queue_config, +# resubscribe=False, +# num_messages_two_subscribers=50, +# num_retain_messages=60, +# ) + diff --git a/test/broker/persist_module_helper.py b/test/broker/persist_module_helper.py index 15b9594a..e6a83e36 100644 --- a/test/broker/persist_module_helper.py +++ b/test/broker/persist_module_helper.py @@ -47,6 +47,7 @@ def publish_messages( topic: str, start: int, end: int, + retain_end=0, message_expiry: int = 0, qos: int = 1, ): @@ -65,6 +66,7 @@ def publish_messages( mid=mid, qos=qos, payload=payload.encode("UTF-8"), + retain = True if i < retain_end else False, proto_ver=proto_ver, properties=props, ) @@ -80,10 +82,11 @@ def check_db( client_msg_counts: dict[str, int], publisher_id: str, num_published_msgs: int, + retain_end: int, message_expiry: int = 0, qos: int = 1, ): - count_list = [v for v in client_msg_counts.values() if v] + [0] + count_list = [v for v in client_msg_counts.values() if v is not None] + [0] num_base_msgs = max(count_list) num_subscriptions = sum(1 for c in client_msg_counts.values() if c is not None) num_client_msgs_out = sum(count_list) @@ -91,7 +94,8 @@ def check_db( port, clients=len(client_msg_counts), client_msgs_out=num_client_msgs_out, - base_msgs=num_base_msgs, + base_msgs=num_base_msgs if num_base_msgs > 0 or retain_end == 0 else 1, + retain_msgs=1 if retain_end > 0 else 0, subscriptions=num_subscriptions, ) @@ -131,7 +135,7 @@ def check_db( mid, port, qos, - retain=0, + retain=1 if i < retain_end else 0, idx=i, ) # Check client msg diff --git a/test/broker/persist_sqlite.py b/test/broker/persist_sqlite.py index c42ea2ec..e7888761 100755 --- a/test/broker/persist_sqlite.py +++ b/test/broker/persist_sqlite.py @@ -360,10 +360,6 @@ def check_client_msg( raise ValueError( "Invalid state %d / %d for message %s" % (row[8], state, msg_id) ) - except ValueError as err: - raise ValueError( - str(err) + f" in client message client_id = {client_id} cmsg_id = {idx}" - ) from err finally: con.close() diff --git a/test/unit/broker/subs_stubs.c b/test/unit/broker/subs_stubs.c index a3280cc5..913e84a6 100644 --- a/test/unit/broker/subs_stubs.c +++ b/test/unit/broker/subs_stubs.c @@ -115,6 +115,11 @@ int retain__init(void) return MOSQ_ERR_SUCCESS; } +void retain__expiry_check(struct mosquitto__retainhier **retainhier) +{ + UNUSED(retainhier); +} + void retain__clean(struct mosquitto__retainhier **retainhier) { UNUSED(retainhier);