Updated versions

LXMF/LXMF.py

@@ -12,7 +12,7 @@ FIELD_ICON_APPEARANCE  = 0x04
 FIELD_FILE_ATTACHMENTS = 0x05
 FIELD_IMAGE            = 0x06
 FIELD_AUDIO            = 0x07
-FIELD_THREAD           = 0x08
+FIELD_THREAD           = 0x08 # Bytes, full thread ID hash
 FIELD_COMMANDS         = 0x09
 FIELD_RESULTS          = 0x0A
 FIELD_GROUP            = 0x0B
@@ -20,6 +20,16 @@ FIELD_TICKET           = 0x0C
 FIELD_EVENT            = 0x0D
 FIELD_RNR_REFS         = 0x0E
 FIELD_RENDERER         = 0x0F
+FIELD_REPLY_TO         = 0x30 # Bytes, full LXMessage.hash
+FIELD_REPLY_QUOTE      = 0x31 # Bytes, quoted content in UTF-8 encoding
+FIELD_REACTION         = 0x40 # Dict, see "Reaction dict indices" below
+FIELD_COMMENT          = 0x41 # Dict, see "Comment dict indices" below
+FIELD_CONTINUATION     = 0x42 # Dict, see "Continuation dict indices" below
+
+# Unallocated fields between 0x00 and 0x80, both included,
+# should be considered reserved for future extensibility
+# For experimental and unstable features, it is recommended
+# to use fields above 0xFF.
 
 # For usecases such as including custom data structures,
 # embedding or encapsulating other data types or protocols
@@ -91,6 +101,30 @@ RENDERER_MICRON        = 0x01
 RENDERER_MARKDOWN      = 0x02
 RENDERER_BBCODE        = 0x03
 
+# Clients choose how to handle reaction content, if at all.
+# While reactions are typically a single unicode emoji or
+# similar, the exact implementation and sanitization is
+# left up to the client. When using the FIELD_REACTION
+# field, the contents is a dict with the following keys:
+REACTION_TO            = 0x00 # Bytes, full LXMessage.hash
+REACTION_CONTENT       = 0x01 # Bytes, the reaction content in UTF-8 encoding
+
+# Clients choose how to handle messages intended as comments
+# for other message, if at all. The actual comment content
+# is carried as the normal LXM content, meaning clients that
+# do not support comments will display them as normal messages.
+# When using the FIELD_COMMENT field, the contents is a dict
+# with the following keys:
+COMMENT_FOR            = 0x00 # Bytes, full LXMessage.hash
+
+# Clients choose how to handle messages that continue earlier
+# messages, if at all. The actual continuation content is
+# carried as the normal LXM content, meaning clients that
+# do not support continuations will display them as normal.
+# When using the FIELD_CONTINUATION field, the contents is a
+# dict with the following keys:
+CONTINUATION_OF        = 0x00 # Bytes, full LXMessage.hash
+
 # Optional propagation node metadata fields. These
 # fields may be highly unstable in allocation and
 # availability until the version 1.0.0 release, so use
@@ -135,8 +169,7 @@ def display_name_from_app_data(app_data=None):
                             return None
 
         # Original announce format
-        else:
-            return app_data.decode("utf-8")
+        else: return app_data.decode("utf-8")
 
 def stamp_cost_from_app_data(app_data=None):
     if app_data == None or app_data == b"": return None
@@ -185,8 +218,8 @@ def pn_stamp_cost_from_app_data(app_data=None):
         if pn_announce_data_is_valid(app_data):
             data = msgpack.unpackb(app_data)
             return data[5][0]
-        else:
-            return None
+        
+        else: return None
 
 def pn_announce_data_is_valid(data):
     try:

LXMF/LXMPeer.py

@@ -328,7 +328,7 @@ class LXMPeer:
                                 self.alive = True
                                 self.last_heard = time.time()
                                 self.sync_backoff = 0
-                                min_accepted_cost = min(0, self.propagation_stamp_cost-self.propagation_stamp_cost_flexibility)
+                                min_accepted_cost = max(0, self.propagation_stamp_cost-self.propagation_stamp_cost_flexibility)
 
                                 RNS.log("Synchronisation link to peer "+RNS.prettyhexrep(self.destination_hash)+" established, preparing sync offer...", RNS.LOG_DEBUG)
                                 unhandled_entries = []
@@ -378,9 +378,13 @@ class LXMPeer:
                                     cumulative_size += lxm_transfer_size
                                     unhandled_ids.append(transient_id)
 
+                                if len(unhandled_ids) == 0:
+                                    RNS.log(f"Sync requested for {self}, but no unhandled messages exist after offer preparation. Sync complete.", RNS.LOG_DEBUG)
+                                    return
+
                                 offer = [self.peering_key[0], unhandled_ids]
 
-                                RNS.log(f"Offering {len(unhandled_ids)} messages to peer {RNS.prettyhexrep(self.destination.hash)} ({RNS.prettysize(len(msgpack.packb(unhandled_ids)))})", RNS.LOG_VERBOSE)
+                                RNS.log(f"Offering {len(unhandled_ids)} messages to peer {RNS.prettyhexrep(self.destination.hash)}", RNS.LOG_VERBOSE)
                                 self.last_offer = unhandled_ids
                                 self.link.request(LXMPeer.OFFER_REQUEST_PATH, offer, response_callback=self.offer_response, failed_callback=self.request_failed)
                                 self.state = LXMPeer.REQUEST_SENT
@@ -404,7 +408,7 @@ class LXMPeer:
             if response == LXMPeer.ERROR_NO_IDENTITY:
                 if self.link != None:
                     RNS.log("Remote peer indicated that no identification was received, retrying...", RNS.LOG_VERBOSE)
-                    self.link.identify()
+                    self.link.identify(self.router.identity)
                     self.state = LXMPeer.LINK_READY
                     self.sync()
                     return

LXMF/LXMRouter.py

@@ -53,6 +53,9 @@ class LXMRouter:
     PROPAGATION_COST_FLEX = 3
     PROPAGATION_COST      = 16
     PROPAGATION_LIMIT     = 256
+    SEQUENTIAL_VALIDATION = True
+    STATIC_SEQUENTIAL     = False
+    MAX_INBOUND_SYNCS     = 3
     SYNC_LIMIT            = PROPAGATION_LIMIT*40
     DELIVERY_LIMIT        = 1000
 
@@ -76,6 +79,11 @@ class LXMRouter:
 
     PR_ALL_MESSAGES       = 0x00
 
+    OFFER_UNKNOWN         = 0x00
+    OFFER_ACCEPTED        = 0x01
+    OFFER_TRANSFERRING    = 0x02
+    OFFER_VALIDATING      = 0x03
+
     DUPLICATE_SIGNAL      = "lxmf_duplicate"
 
     STATS_GET_PATH        = "/pn/get/stats"
@@ -86,12 +94,13 @@ class LXMRouter:
     ### Developer-facing API ##############################
     #######################################################
 
-    def __init__(self, identity=None, storagepath=None, autopeer=AUTOPEER, autopeer_maxdepth=None,
+    def __init__(self, identity=None, storagepath=None, name=None, autopeer=AUTOPEER, autopeer_maxdepth=None,
                  propagation_limit=PROPAGATION_LIMIT, delivery_limit=DELIVERY_LIMIT, sync_limit=SYNC_LIMIT,
                  enforce_ratchets=False, enforce_stamps=False, static_peers = [], max_peers=None,
                  from_static_only=False, sync_strategy=LXMPeer.STRATEGY_PERSISTENT,
                  propagation_cost=PROPAGATION_COST, propagation_cost_flexibility=PROPAGATION_COST_FLEX,
-                 peering_cost=PEERING_COST, max_peering_cost=MAX_PEERING_COST, name=None):
+                 peering_cost=PEERING_COST, max_peering_cost=MAX_PEERING_COST, max_inbound_syncs=MAX_INBOUND_SYNCS,
+                 sequential_validation=SEQUENTIAL_VALIDATION, static_sequential=STATIC_SEQUENTIAL):
 
         random.seed(os.urandom(10))
 
@@ -131,6 +140,9 @@ class LXMRouter:
         self.information_storage_limit          = None
         self.propagation_per_transfer_limit     = propagation_limit
         self.propagation_per_sync_limit         = sync_limit
+        self.propagation_sequential_validation  = sequential_validation
+        self.propagation_static_peer_sequential = static_sequential
+        self.propagation_max_inbound_syncs      = max_inbound_syncs
         self.delivery_per_transfer_limit        = delivery_limit
         self.propagation_stamp_cost             = propagation_cost
         self.propagation_stamp_cost_flexibility = propagation_cost_flexibility
@@ -148,11 +160,13 @@ class LXMRouter:
         self.wants_download_on_path_available_to = None
         self.propagation_transfer_state = LXMRouter.PR_IDLE
         self.propagation_transfer_progress = 0.0
+        self.propagation_transfer_size = None
         self.propagation_transfer_last_result = None
         self.propagation_transfer_last_duplicates = None
         self.propagation_transfer_max_messages = None
         self.prioritise_rotating_unreachable_peers = False
         self.active_propagation_links = []
+        self.accepted_offer_links = {}
         self.validated_peer_links = {}
         self.locally_delivered_transient_ids = {}
         self.locally_processed_transient_ids = {}
@@ -160,18 +174,24 @@ class LXMRouter:
         self.available_tickets = {"outbound": {}, "inbound": {}, "last_deliveries": {}}
 
         self.outbound_processing_lock = threading.Lock()
+        self.delivered_transient_ids_lock = threading.Lock()
+        self.processed_transient_ids_lock = threading.Lock()
         self.cost_file_lock = threading.Lock()
         self.ticket_file_lock = threading.Lock()
         self.stamp_gen_lock = threading.Lock()
+        self.accepted_offer_links_lock = threading.Lock()
+        self.sequential_validation_lock = threading.Lock()
+        self.incoming_delivery_resource_lock = threading.Lock()
         self.exit_handler_running = False
 
-        if identity == None:
-            identity = RNS.Identity()
+        if identity == None: identity = RNS.Identity()
 
         self.identity = identity
         self.propagation_destination = RNS.Destination(self.identity, RNS.Destination.IN, RNS.Destination.SINGLE, APP_NAME, "propagation")
         self.propagation_destination.set_default_app_data(self.get_propagation_node_app_data)
         self.control_destination = None
+        self.validating_pn_stamps_from = {}
+        self.incoming_delivery_resources = {}
         self.client_propagation_messages_received = 0
         self.client_propagation_messages_served = 0
         self.unpeered_propagation_incoming = 0
@@ -237,9 +257,7 @@ class LXMRouter:
             RNS.log("Could not load locally processed message ID cache from storage. The contained exception was: "+str(e), RNS.LOG_ERROR)
             self.locally_processed_transient_ids = {}
 
-        try:
-            self.clean_transient_id_caches()
-
+        try: self.clean_transient_id_caches()
         except Exception as e:
             RNS.log("Could not clean transient ID caches. The contained exception was : "+str(e), RNS.LOG_ERROR)
             self.locally_delivered_transient_ids = {}
@@ -486,6 +504,7 @@ class LXMRouter:
             max_messages = LXMRouter.PR_ALL_MESSAGES
 
         self.propagation_transfer_progress = 0.0
+        self.propagation_transfer_size = None
         self.propagation_transfer_max_messages = max_messages
         if self.outbound_propagation_node != None:
             if self.outbound_propagation_link != None and self.outbound_propagation_link.status == RNS.Link.ACTIVE:
@@ -852,6 +871,7 @@ class LXMRouter:
     JOB_OUTBOUND_INTERVAL  = 1
     JOB_STAMPS_INTERVAL    = 1
     JOB_LINKS_INTERVAL     = 1
+    JOB_RESOURCE_INTERVAL  = 2
     JOB_TRANSIENT_INTERVAL = 60
     JOB_STORE_INTERVAL     = 120
     JOB_PEERSYNC_INTERVAL  = 6
@@ -870,6 +890,9 @@ class LXMRouter:
             if self.processing_count % LXMRouter.JOB_LINKS_INTERVAL == 0:
                 self.clean_links()
 
+            if self.processing_count % LXMRouter.JOB_RESOURCE_INTERVAL == 0:
+                self.clean_resource_tracking()
+
             if self.processing_count % LXMRouter.JOB_TRANSIENT_INTERVAL == 0:
                 self.clean_transient_id_caches()
 
@@ -890,8 +913,7 @@ class LXMRouter:
         while (True):
             # TODO: Improve this to scheduling, so manual
             # triggers can delay next run
-            try:
-                self.jobs()
+            try: self.jobs()
             except Exception as e:
                 RNS.log("An error ocurred while running LXMF Router jobs.", RNS.LOG_ERROR)
                 RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR)
@@ -910,6 +932,22 @@ class LXMRouter:
 
             RNS.log(f"Distribution queue mapping completed in {RNS.prettytime(time.time()-st)}", RNS.LOG_DEBUG)
 
+    def clean_resource_tracking(self):
+        try:
+            stale_resources = []
+            with self.incoming_delivery_resource_lock:
+                for resource_hash in self.incoming_delivery_resources:
+                    if self.incoming_delivery_resources[resource_hash].status >= RNS.Resource.COMPLETE:
+                        stale_resources.append(resource_hash)
+
+                for resource_hash in stale_resources: self.incoming_delivery_resources.pop(resource_hash)
+                cleaned = len(stale_resources)
+                if cleaned > 0: RNS.log(f"Cleaned {cleaned} resource{'s' if cleaned != 1 else ''} from inbound tracking", RNS.LOG_DEBUG)
+
+        except Exception as e:
+            RNS.log(f"Error while cleaning incoming delivery resource tracking: {e}", RNS.LOG_ERROR)
+            RNS.trace_exception(e)
+
     def clean_links(self):
         closed_links = []
         for link_hash in self.direct_links:
@@ -936,6 +974,17 @@ class LXMRouter:
                 self.active_propagation_links.remove(link)
                 link.teardown()
 
+            active_link_ids = []
+            inactive_offers = []
+            for link in self.active_propagation_links: active_link_ids.append(link.link_id)
+            with self.accepted_offer_links_lock:
+                for link_id in self.accepted_offer_links:
+                    if not link_id in active_link_ids: inactive_offers.append(link_id)
+
+                for link_id in inactive_offers:
+                    RNS.log(f"Cleaning inbound sync link accounting for link {RNS.prettyhexrep(link_id)} since link is no longer active", RNS.LOG_DEBUG) # TODO: Remove at some point
+                    self.accepted_offer_links.pop(link_id)
+        
         except Exception as e:
             RNS.log("An error occurred while cleaning inbound propagation links. The contained exception was: "+str(e), RNS.LOG_ERROR)
 
@@ -956,27 +1005,27 @@ class LXMRouter:
     def clean_transient_id_caches(self):
         now = time.time()
         removed_entries = []
-        for transient_id in self.locally_delivered_transient_ids:
-            timestamp = self.locally_delivered_transient_ids[transient_id]
-            if now > timestamp+LXMRouter.MESSAGE_EXPIRY*6.0:
-                removed_entries.append(transient_id)
+        for transient_id in self.locally_delivered_transient_ids.copy():
+            timestamp = None
+            with self.delivered_transient_ids_lock: timestamp = self.locally_delivered_transient_ids[transient_id]
+            if timestamp and now > timestamp+LXMRouter.MESSAGE_EXPIRY*6.0: removed_entries.append(transient_id)
 
         for transient_id in removed_entries:
-            self.locally_delivered_transient_ids.pop(transient_id)
-            RNS.log("Cleaned "+RNS.prettyhexrep(transient_id)+" from local delivery cache", RNS.LOG_DEBUG)
+            with self.delivered_transient_ids_lock: self.locally_delivered_transient_ids.pop(transient_id)
+            RNS.log("Cleaned "+RNS.prettyhexrep(transient_id)+" from local delivery cache", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
 
         removed_entries = []
         for transient_id in self.locally_processed_transient_ids:
-            timestamp = self.locally_processed_transient_ids[transient_id]
-            if now > timestamp+LXMRouter.MESSAGE_EXPIRY*6.0:
-                removed_entries.append(transient_id)
+            timestampt = None
+            with self.processed_transient_ids_lock: timestamp = self.locally_processed_transient_ids[transient_id]
+            if timestamp and now > timestamp+LXMRouter.MESSAGE_EXPIRY*6.0: removed_entries.append(transient_id)
 
         for transient_id in removed_entries:
-            self.locally_processed_transient_ids.pop(transient_id)
-            RNS.log("Cleaned "+RNS.prettyhexrep(transient_id)+" from locally processed cache", RNS.LOG_DEBUG)
+            with self.processed_transient_ids_lock: self.locally_processed_transient_ids.pop(transient_id)
+            RNS.log("Cleaned "+RNS.prettyhexrep(transient_id)+" from locally processed cache", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
 
     def update_stamp_cost(self, destination_hash, stamp_cost):
-        RNS.log(f"Updating outbound stamp cost for {RNS.prettyhexrep(destination_hash)} to {stamp_cost}", RNS.LOG_DEBUG)
+        RNS.log(f"Updating outbound stamp cost for {RNS.prettyhexrep(destination_hash)} to {stamp_cost}", RNS.LOG_PATHING) if RNS.sl(RNS.LOG_PATHING) else None
         self.outbound_stamp_costs[destination_hash] = [time.time(), stamp_cost]
         
         def job(): self.save_outbound_stamp_costs()
@@ -996,7 +1045,7 @@ class LXMRouter:
                     stamp_cost = delivery_destination.stamp_cost
 
             supported_functionality = [SF_COMPRESSION]
-            peer_data = [display_name, stamp_cost]
+            peer_data = [display_name, stamp_cost, supported_functionality]
 
             return msgpack.packb(peer_data)
 
@@ -1177,11 +1226,12 @@ class LXMRouter:
     def save_locally_delivered_transient_ids(self):
         try:
             if len(self.locally_delivered_transient_ids) > 0:
-                if not os.path.isdir(self.storagepath):
-                    os.makedirs(self.storagepath)
-
-                with open(self.storagepath+"/local_deliveries", "wb") as locally_delivered_file:
-                    locally_delivered_file.write(msgpack.packb(self.locally_delivered_transient_ids))
+                if not os.path.isdir(self.storagepath): os.makedirs(self.storagepath)
+                write_path = self.storagepath+"/local_deliveries"
+                temp_path  = write_path+".tmp."+str(time.time())
+                with open(temp_path, "wb") as locally_delivered_file:
+                    locally_delivered_file.write(msgpack.packb(self.locally_delivered_transient_ids.copy()))
+                os.replace(temp_path, write_path)
 
         except Exception as e:
             RNS.log("Could not save locally delivered message ID cache to storage. The contained exception was: "+str(e), RNS.LOG_ERROR)
@@ -1192,8 +1242,11 @@ class LXMRouter:
                 if not os.path.isdir(self.storagepath):
                     os.makedirs(self.storagepath)
 
-                with open(self.storagepath+"/locally_processed", "wb") as locally_processed_file:
-                    locally_processed_file.write(msgpack.packb(self.locally_processed_transient_ids))
+                write_path = self.storagepath+"/locally_processed"
+                temp_path  = write_path+".tmp."+str(time.time())
+                with open(temp_path, "wb") as locally_processed_file:
+                    locally_processed_file.write(msgpack.packb(self.locally_processed_transient_ids.copy()))
+                os.replace(temp_path, write_path)
 
         except Exception as e:
             RNS.log("Could not save locally processed transient ID cache to storage. The contained exception was: "+str(e), RNS.LOG_ERROR)
@@ -1203,19 +1256,19 @@ class LXMRouter:
             if not os.path.isdir(self.storagepath):
                 os.makedirs(self.storagepath)
 
-            with open(self.storagepath+"/node_stats", "wb") as stats_file:
-                node_stats = {
-                    "client_propagation_messages_received": self.client_propagation_messages_received,
-                    "client_propagation_messages_served": self.client_propagation_messages_served,
-                    "unpeered_propagation_incoming": self.unpeered_propagation_incoming,
-                    "unpeered_propagation_rx_bytes": self.unpeered_propagation_rx_bytes,
-                }
+            write_path = self.storagepath+"/node_stats"
+            temp_path  = write_path+".tmp."+str(time.time())
+            with open(temp_path, "wb") as stats_file:
+                node_stats = {"client_propagation_messages_received": self.client_propagation_messages_received,
+                              "client_propagation_messages_served": self.client_propagation_messages_served,
+                              "unpeered_propagation_incoming": self.unpeered_propagation_incoming,
+                              "unpeered_propagation_rx_bytes": self.unpeered_propagation_rx_bytes}
                 stats_file.write(msgpack.packb(node_stats))
+            os.replace(temp_path, write_path)
 
         except Exception as e:
             RNS.log("Could not save local node stats to storage. The contained exception was: "+str(e), RNS.LOG_ERROR)
 
-
     def clean_outbound_stamp_costs(self):
         try:
             expired = []
@@ -1234,12 +1287,13 @@ class LXMRouter:
     def save_outbound_stamp_costs(self):
         with self.cost_file_lock:
             try:
-                if not os.path.isdir(self.storagepath):
-                        os.makedirs(self.storagepath)
+                if not os.path.isdir(self.storagepath): os.makedirs(self.storagepath)
 
-                outbound_stamp_costs_file = open(self.storagepath+"/outbound_stamp_costs", "wb")
-                outbound_stamp_costs_file.write(msgpack.packb(self.outbound_stamp_costs.copy()))
-                outbound_stamp_costs_file.close()
+                write_path = self.storagepath+"/outbound_stamp_costs"
+                temp_path  = write_path+".tmp."+str(time.time())
+                with open(temp_path, "wb") as outbound_stamp_costs_file:
+                    outbound_stamp_costs_file.write(msgpack.packb(self.outbound_stamp_costs.copy()))
+                os.replace(temp_path, write_path)
 
             except Exception as e:
                 RNS.log("Could not save outbound stamp costs to storage. The contained exception was: "+str(e), RNS.LOG_ERROR)
@@ -1278,9 +1332,11 @@ class LXMRouter:
                 if not os.path.isdir(self.storagepath):
                         os.makedirs(self.storagepath)
 
-                available_tickets_file = open(self.storagepath+"/available_tickets", "wb")
-                available_tickets_file.write(msgpack.packb(self.available_tickets))
-                available_tickets_file.close()
+                write_path = self.storagepath+"/available_tickets"
+                temp_path  = write_path+".tmp."+str(time.time())
+                with open(temp_path, "wb") as available_tickets_file:
+                    available_tickets_file.write(msgpack.packb(self.available_tickets))
+                os.replace(temp_path, write_path)
 
             except Exception as e:
                 RNS.log("Could not save available tickets to storage. The contained exception was: "+str(e), RNS.LOG_ERROR)
@@ -1337,10 +1393,8 @@ class LXMRouter:
             self.propagation_destination.deregister_request_handler(LXMRouter.UNPEER_REQUEST_PATH)
             for link in self.active_propagation_links:
                 try:
-                    if link.status == RNS.Link.ACTIVE:
-                        link.teardown()
-                except Exception as e:
-                    RNS.log("Error while tearing down propagation link: {e}", RNS.LOG_ERROR)
+                    if link.status == RNS.Link.ACTIVE: link.teardown()
+                except Exception as e: RNS.log("Error while tearing down propagation link: {e}", RNS.LOG_ERROR)
 
         RNS.log("Persisting LXMF state data to storage...", RNS.LOG_NOTICE)
         self.flush_queues()
@@ -1353,9 +1407,11 @@ class LXMRouter:
                     peer = self.peers[peer_id]
                     serialised_peers.append(peer.to_bytes())
 
-                peers_file = open(self.storagepath+"/peers", "wb")
-                peers_file.write(msgpack.packb(serialised_peers))
-                peers_file.close()
+                write_path = self.storagepath+"/peers"
+                temp_path  = write_path+".tmp."+str(time.time())
+                with open(temp_path, "wb") as peers_file:
+                    peers_file.write(msgpack.packb(serialised_peers))
+                os.replace(temp_path, write_path)
 
                 RNS.log(f"Saved {len(serialised_peers)} peers to storage in {RNS.prettyshorttime(time.time()-st)}", RNS.LOG_NOTICE)
 
@@ -1590,6 +1646,7 @@ class LXMRouter:
     def message_get_progress(self, request_receipt):
         self.propagation_transfer_state = LXMRouter.PR_RECEIVING
         self.propagation_transfer_progress = request_receipt.get_progress()
+        if request_receipt.response_size: self.propagation_transfer_size = request_receipt.response_size
 
     def message_get_failed(self, request_receipt):
         RNS.log("Message list/get request failed", RNS.LOG_DEBUG)
@@ -1599,21 +1656,66 @@ class LXMRouter:
     def acknowledge_sync_completion(self, reset_state=False, failure_state=None):
         self.propagation_transfer_last_result = None
         if reset_state or self.propagation_transfer_state <= LXMRouter.PR_COMPLETE:
-            if failure_state == None:
-                self.propagation_transfer_state = LXMRouter.PR_IDLE
-            else:
-                self.propagation_transfer_state = failure_state
+            if failure_state == None: self.propagation_transfer_state = LXMRouter.PR_IDLE
+            else:                     self.propagation_transfer_state = failure_state
 
         self.propagation_transfer_progress = 0.0
+        self.propagation_transfer_size = None
         self.wants_download_on_path_available_from = None
         self.wants_download_on_path_available_to = None
 
     def has_message(self, transient_id):
-        if transient_id in self.locally_delivered_transient_ids:
-            return True
-        else:
+        if transient_id in self.locally_delivered_transient_ids: return True
+        else:                                                    return False
+    
+    def inbound_count(self):
+        try:
+            with self.incoming_delivery_resource_lock:
+                return len([r for r in self.incoming_delivery_resources if self.incoming_delivery_resources[r].status < RNS.Resource.COMPLETE])
+        except Exception as e:
+            RNS.log(f"Error while getting inbound resource transfer count: {e}", RNS.LOG_ERROR)
+            return 0
+
+    def inbound_resources(self):
+        active_resources = []
+        with self.incoming_delivery_resource_lock:
+            for resource_hash in self.incoming_delivery_resources:
+                resource = self.incoming_delivery_resources[resource_hash]
+                if resource.status < RNS.Resource.COMPLETE:
+                    active_resources.append(resource)
+
+        return active_resources
+
+    def cancel_inbound(self, resource_hash):
+        resource = None
+        with self.incoming_delivery_resource_lock:
+            if resource_hash in self.incoming_delivery_resources:
+                resource = self.incoming_delivery_resources[resource_hash]
+
+        if not resource:
+            RNS.log(f"Resource {RNS.prettyhexrep(resource_hash)} not found, cannot cancel", RNS.LOG_WARNING)
             return False
 
+        else:
+            if resource.status < RNS.Resource.COMPLETE:
+                resource.cancel()
+                RNS.log(f"Cancelled incoming delivery resource {resource}", RNS.LOG_NOTICE)
+                return True
+            else:
+                RNS.log(f"Incoming delivery resource {resource} already concluded, cannot cancel", RNS.LOG_WARNING)
+                return False
+
+    def cancel_all_inbound(self):
+        active_resources = []
+        with self.incoming_delivery_resource_lock:
+            for resource_hash in self.incoming_delivery_resources:
+                resource = self.incoming_delivery_resources[resource_hash]
+                if resource.status < RNS.Resource.COMPLETE:
+                    active_resources.append(resource)
+
+        for resource in active_resources: resource.cancel()
+        return len(active_resources)
+
     def cancel_outbound(self, message_id, cancel_state=LXMessage.CANCELLED):
         try:
             if message_id in self.pending_deferred_stamps:
@@ -1735,11 +1837,13 @@ class LXMRouter:
     def lxmf_delivery(self, lxmf_data, destination_type = None, phy_stats = None, ratchet_id = None, method = None, no_stamp_enforcement=False, allow_duplicate=False):
         try:
             message = LXMessage.unpack_from_bytes(lxmf_data)
-            if ratchet_id and not message.ratchet_id:
-                message.ratchet_id = ratchet_id
 
-            if method:
-                message.method = method
+            if message.source_blackholed:
+                RNS.log(f"Dropping LXM from blackholed identity {message.source.identity}", RNS.LOG_DEBUG)
+                return False
+
+            if ratchet_id and not message.ratchet_id: message.ratchet_id = ratchet_id
+            if method: message.method = method
 
             if message.signature_validated and FIELD_TICKET in message.fields:
                 ticket_entry = message.fields[FIELD_TICKET]
@@ -1803,11 +1907,11 @@ class LXMRouter:
                 RNS.log(str(self)+" ignored already received message from "+RNS.prettyhexrep(message.source_hash), RNS.LOG_DEBUG)
                 return False
             else:
-                self.locally_delivered_transient_ids[message.hash] = time.time()
+                with self.delivered_transient_ids_lock:
+                    self.locally_delivered_transient_ids[message.hash] = time.time()
 
             if self.__delivery_callback != None and callable(self.__delivery_callback):
-                try:
-                    self.__delivery_callback(message)
+                try: self.__delivery_callback(message)
                 except Exception as e:
                     RNS.log("An error occurred in the external delivery callback for "+str(message), RNS.LOG_ERROR)
                     RNS.trace_exception(e)
@@ -1854,15 +1958,21 @@ class LXMRouter:
         link.set_packet_callback(self.delivery_packet)
         link.set_resource_strategy(RNS.Link.ACCEPT_APP)
         link.set_resource_callback(self.delivery_resource_advertised)
-        link.set_resource_started_callback(self.resource_transfer_began)
+        link.set_resource_started_callback(self.delivery_resource_transfer_began)
         link.set_resource_concluded_callback(self.delivery_resource_concluded)
         link.set_remote_identified_callback(self.delivery_remote_identified)
 
     def delivery_link_closed(self, link):
         pass
 
-    def resource_transfer_began(self, resource):
-        RNS.log("Transfer began for LXMF delivery resource "+str(resource), RNS.LOG_DEBUG)
+    def delivery_resource_transfer_began(self, resource):
+        size = resource.get_data_size()
+        with self.incoming_delivery_resource_lock: self.incoming_delivery_resources[resource.hash] = resource
+        RNS.log(f"Began {RNS.prettysize(size) if size else 'unknown size'} transfer for LXMF delivery resource {resource}", RNS.LOG_DEBUG)
+
+    def propagation_resource_transfer_began(self, resource):
+        size = resource.get_data_size()
+        RNS.log(f"Began {RNS.prettysize(size) if size else 'unknown size'} transfer for LXMF propagation resource {resource}", RNS.LOG_DEBUG)
 
     def delivery_resource_advertised(self, resource):
         size = resource.get_data_size()
@@ -1878,8 +1988,7 @@ class LXMRouter:
         if resource.status == RNS.Resource.COMPLETE:
             ratchet_id = None
             # Set ratchet ID to link ID if available
-            if resource.link and hasattr(resource.link, "link_id"):
-                ratchet_id = resource.link.link_id
+            if resource.link and hasattr(resource.link, "link_id"): ratchet_id = resource.link.link_id
             phy_stats = {"rssi": resource.link.rssi, "snr": resource.link.snr, "q": resource.link.q}
             self.lxmf_delivery(resource.data.read(), resource.link.type, phy_stats=phy_stats, ratchet_id=ratchet_id, method=LXMessage.DIRECT)
 
@@ -1986,19 +2095,15 @@ class LXMRouter:
                                 # Don't consider for unpeering until at
                                 # least one message has been offered
                                 pass
-                            else:
-                                waiting_peers.append(peer)
-                        else:
-                            unresponsive_peers.append(peer)
+                            else: waiting_peers.append(peer)
+                        else: unresponsive_peers.append(peer)
 
                 drop_pool = []
                 if len(unresponsive_peers) > 0:
                     drop_pool.extend(unresponsive_peers)
                     if not self.prioritise_rotating_unreachable_peers:
                         drop_pool.extend(waiting_peers)
-
-                else:
-                    drop_pool.extend(waiting_peers)
+                else: drop_pool.extend(waiting_peers)
 
                 if len(drop_pool) > 0:
                     drop_count = min(required_drops, len(drop_pool))
@@ -2084,10 +2189,20 @@ class LXMRouter:
         link.set_packet_callback(self.propagation_packet)
         link.set_resource_strategy(RNS.Link.ACCEPT_APP)
         link.set_resource_callback(self.propagation_resource_advertised)
-        link.set_resource_started_callback(self.resource_transfer_began)
+        link.set_resource_started_callback(self.propagation_resource_transfer_began)
         link.set_resource_concluded_callback(self.propagation_resource_concluded)
         self.active_propagation_links.append(link)
 
+    @property
+    def propagation_resources_transferring(self):
+        count = 0
+        with self.accepted_offer_links_lock:
+            for link_id in self.accepted_offer_links:
+                if self.accepted_offer_links[link_id] > self.OFFER_ACCEPTED:
+                    count += 1
+
+        return count
+
     def propagation_resource_advertised(self, resource):
         if self.from_static_only:
             remote_identity = resource.link.get_remote_identity()
@@ -2107,7 +2222,13 @@ class LXMRouter:
         if limit != None and size > limit:
             RNS.log(f"Rejecting {RNS.prettysize(size)} incoming propagation resource, since it exceeds the limit of {RNS.prettysize(limit)}", RNS.LOG_DEBUG)
             return False
+
         else:
+            with self.accepted_offer_links_lock:
+                if resource.link.link_id in self.accepted_offer_links:
+                    ri_str = RNS.prettyhexrep(resource.link.get_remote_identity().hash) if resource.link.get_remote_identity() else 'unknown peer'
+                    RNS.log(f"Sync offer for {ri_str} started transferring", RNS.LOG_DEBUG) # TODO: Remove at some point
+                    self.accepted_offer_links[resource.link.link_id] = self.OFFER_TRANSFERRING
             return True
 
     def propagation_packet(self, data, packet):
@@ -2143,13 +2264,24 @@ class LXMRouter:
             RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR)
 
     def offer_request(self, path, data, request_id, link_id, remote_identity, requested_at):
-        if remote_identity == None:
-            return LXMPeer.ERROR_NO_IDENTITY
+        if remote_identity == None: return LXMPeer.ERROR_NO_IDENTITY
         else:
             remote_destination = RNS.Destination(remote_identity, RNS.Destination.OUT, RNS.Destination.SINGLE, APP_NAME, "propagation")
             remote_hash = remote_destination.hash
             remote_str  = RNS.prettyhexrep(remote_hash)
 
+            bypass_sequential = not self.propagation_static_peer_sequential and remote_hash in self.static_peers
+            if not bypass_sequential and self.propagation_sequential_validation and len(self.validating_pn_stamps_from) > 0:
+                RNS.log(f"Propagation offer from node {remote_str} postponed, already validating {len(self.validating_pn_stamps_from)} PN stamp batches", RNS.LOG_NOTICE)
+                # for rh in self.validating_pn_stamps_from:
+                #     RNS.log(f"Validating from {RNS.prettyhexrep(rh)} for {RNS.prettytime(time.time()-self.validating_pn_stamps_from[rh])}")
+                return LXMPeer.ERROR_THROTTLED
+
+            resources_transferring = self.propagation_resources_transferring
+            if not bypass_sequential and self.propagation_max_inbound_syncs and resources_transferring >= self.propagation_max_inbound_syncs:
+                RNS.log(f"Propagation offer from node {remote_str} postponed, already receiving {resources_transferring} sync resource{'s' if resources_transferring != 1 else ''}", RNS.LOG_NOTICE)
+                return LXMPeer.ERROR_THROTTLED
+
             if remote_hash in self.throttled_peers:
                 throttle_remaining = self.throttled_peers[remote_hash]-time.time()
                 if throttle_remaining > 0:
@@ -2185,9 +2317,16 @@ class LXMRouter:
                     for transient_id in transient_ids:
                         if not transient_id in self.propagation_entries: wanted_ids.append(transient_id)
 
-                    if len(wanted_ids)   == 0:                  return False
-                    elif len(wanted_ids) == len(transient_ids): return True
-                    else:                                       return wanted_ids
+                    if len(wanted_ids) == 0:
+                        RNS.log(f"No wanted messages in offer from {RNS.prettyhexrep(remote_hash)}", RNS.LOG_DEBUG)
+                        return False
+                    elif len(wanted_ids) == len(transient_ids):
+                        RNS.log(f"Accepted all {len(wanted_ids)} offered message{'s' if len(wanted_ids) != 1 else ''} from {RNS.prettyhexrep(remote_hash)}", RNS.LOG_DEBUG)
+                        return True
+                    else:
+                        RNS.log(f"Accepted {len(wanted_ids)} offered message{'s' if len(wanted_ids) != 1 else ''} from {RNS.prettyhexrep(remote_hash)}", RNS.LOG_DEBUG)
+                        with self.accepted_offer_links_lock: self.accepted_offer_links[link_id] = self.OFFER_ACCEPTED
+                        return wanted_ids
 
             except Exception as e:
                 RNS.log("Error occurred while generating response for sync request, the contained exception was: "+str(e), RNS.LOG_DEBUG)
@@ -2248,12 +2387,41 @@ class LXMRouter:
                         ms = "" if len(messages) == 1 else "s"
                         RNS.log(f"Received {len(messages)} message{ms} from {remote_str}, validating stamps...", RNS.LOG_VERBOSE)
 
-                        min_accepted_cost  = max(0, self.propagation_stamp_cost-self.propagation_stamp_cost_flexibility)
-                        validated_messages = LXStamper.validate_pn_stamps(messages, min_accepted_cost)
-                        invalid_stamps     = len(messages)-len(validated_messages)
-                        ms                 = "" if invalid_stamps == 1 else "s"
-                        if len(validated_messages) == len(messages): RNS.log(f"All message stamps validated from {remote_str}", RNS.LOG_VERBOSE)
-                        else:                                        RNS.log(f"Transfer from {remote_str} contained {invalid_stamps} invalid stamp{ms}", RNS.LOG_WARNING)
+                        with self.accepted_offer_links_lock:
+                            if remote_hash:
+                                RNS.log(f"Updating sync link accounting entry for {RNS.prettyhexrep(remote_hash)} to validating", RNS.LOG_DEBUG) # TODO: Remove at some point
+                                self.accepted_offer_links[resource.link.link_id] = self.OFFER_VALIDATING
+
+                        with self.sequential_validation_lock:
+                            if remote_hash:
+                                RNS.log(f"Adding validation job accounting entry for {RNS.prettyhexrep(remote_hash)}", RNS.LOG_DEBUG) # TODO: Remove at some point
+                                self.validating_pn_stamps_from[remote_hash] = time.time()
+
+                        try:
+                            min_accepted_cost  = max(0, self.propagation_stamp_cost-self.propagation_stamp_cost_flexibility)
+                            validated_messages = LXStamper.validate_pn_stamps(messages, min_accepted_cost)
+                            invalid_stamps     = len(messages)-len(validated_messages)
+                            ms                 = "" if invalid_stamps == 1 else "s"
+                            if len(validated_messages) == len(messages): RNS.log(f"All message stamps validated from {remote_str}", RNS.LOG_VERBOSE)
+                            else:                                        RNS.log(f"Transfer from {remote_str} contained {invalid_stamps} invalid stamp{ms}", RNS.LOG_WARNING)
+
+                        except Exception as e:
+                            RNS.log(f"Error while validating received propagation message stamps: {e}", RNS.LOG_ERROR)
+                            RNS.trace_exception(e)
+                            return
+
+                        finally:
+                            if remote_hash:
+                                with self.sequential_validation_lock:
+                                    RNS.log(f"Cleaning validation job accounting entry for {RNS.prettyhexrep(remote_hash)}", RNS.LOG_DEBUG) # TODO: Remove at some point
+                                    try: self.validating_pn_stamps_from.pop(remote_hash)
+                                    except Exception as e: RNS.log(f"Failed to remove PN stamp validation job from sequential tracking: {e}", RNS.LOG_ERROR)
+
+                            with self.accepted_offer_links_lock:
+                                if resource.link.link_id in self.accepted_offer_links:
+                                    ri_str = RNS.prettyhexrep(resource.link.get_remote_identity().hash) if resource.link.get_remote_identity() else 'unknown peer'
+                                    RNS.log(f"Cleaning inbound sync link accounting for {ri_str}", RNS.LOG_DEBUG) # TODO: Remove at some point
+                                    self.accepted_offer_links.pop(resource.link.link_id)
 
                         for validated_entry in validated_messages:
                             transient_id = validated_entry[0]
@@ -2292,6 +2460,12 @@ class LXMRouter:
                 RNS.log("Error while unpacking received propagation resource", RNS.LOG_DEBUG)
                 RNS.trace_exception(e)
 
+        with self.accepted_offer_links_lock:
+            if resource.link.link_id in self.accepted_offer_links:
+                ri_str = RNS.prettyhexrep(resource.link.get_remote_identity().hash) if resource.link.get_remote_identity() else 'unknown peer'
+                RNS.log(f"Cleaning inbound sync link accounting for {ri_str} on resource failure", RNS.LOG_DEBUG) # TODO: Remove at some point
+                self.accepted_offer_links.pop(resource.link.link_id)
+
     def enqueue_peer_distribution(self, transient_id, from_peer):
         self.peer_distribution_queue.append([transient_id, from_peer])
 
@@ -2322,8 +2496,7 @@ class LXMRouter:
                 if (not transient_id in self.propagation_entries and not transient_id in self.locally_processed_transient_ids) or allow_duplicate == True:
                     received = time.time()
                     destination_hash  = lxmf_data[:LXMessage.DESTINATION_LENGTH]
-
-                    self.locally_processed_transient_ids[transient_id] = received
+                    with self.processed_transient_ids_lock: self.locally_processed_transient_ids[transient_id] = received
 
                     if destination_hash in self.delivery_destinations:
                         delivery_destination = self.delivery_destinations[destination_hash]
@@ -2332,18 +2505,14 @@ class LXMRouter:
                         if decrypted_lxmf_data != None:
                             delivery_data = lxmf_data[:LXMessage.DESTINATION_LENGTH]+decrypted_lxmf_data
                             self.lxmf_delivery(delivery_data, delivery_destination.type, ratchet_id=delivery_destination.latest_ratchet_id, method=LXMessage.PROPAGATED, no_stamp_enforcement=no_stamp_enforcement, allow_duplicate=allow_duplicate)
-                            self.locally_delivered_transient_ids[transient_id] = time.time()
-
-                            if signal_local_delivery != None:
-                                return signal_local_delivery
-
+                            with self.delivered_transient_ids_lock: self.locally_delivered_transient_ids[transient_id] = time.time()
+                            if signal_local_delivery != None: return signal_local_delivery
                     else:
                         if self.propagation_node:
                             stamped_data    = lxmf_data+stamp_data
                             value_component = f"_{stamp_value}" if stamp_value and stamp_value > 0 else ""
                             file_path       = f"{self.messagepath}/{RNS.hexrep(transient_id, delimit=False)}_{received}{value_component}"
-                            msg_file        = open(file_path, "wb")
-                            msg_file.write(stamped_data); msg_file.close()
+                            with open(file_path, "wb") as msg_file: msg_file.write(stamped_data)
 
                             RNS.log(f"Received propagated LXMF message {RNS.prettyhexrep(transient_id)} with stamp value {stamp_value}, adding to peer distribution queues...", RNS.LOG_EXTREME)
                             self.propagation_entries[transient_id] = [destination_hash, file_path, time.time(), len(stamped_data), [], [], stamp_value]
@@ -2404,8 +2573,7 @@ class LXMRouter:
     def process_deferred_stamps(self):
         if len(self.pending_deferred_stamps) > 0:
 
-            if self.stamp_gen_lock.locked():
-                return
+            if self.stamp_gen_lock.locked(): return
 
             else:
                 with self.stamp_gen_lock:
@@ -2414,6 +2582,7 @@ class LXMRouter:
                     for message_id in self.pending_deferred_stamps:
                         lxmessage = self.pending_deferred_stamps[message_id]
                         if selected_lxm == None:
+                            # TODO: Improve logic and add stamp_cost_known here
                             selected_lxm = lxmessage
                             selected_message_id = message_id
 

LXMF/LXMessage.py

@@ -8,6 +8,7 @@ import multiprocessing
 
 import LXMF.LXStamper as LXStamper
 from .LXMF import APP_NAME, compression_support_from_app_data
+from threading import Lock
 
 
 class LXMessage:
@@ -168,6 +169,7 @@ class LXMessage:
         self.paper_packed            = None
 
         self.incoming                = False
+        self.source_blackholed       = False
         self.signature_validated     = False
         self.unverified_reason       = None
         self.ratchet_id              = None
@@ -183,6 +185,7 @@ class LXMessage:
         self.__delivery_destination  = None
         self.__delivery_callback     = None
         self.__pn_encrypted_data     = None
+        self.__persist_lock          = Lock()
         self.failed_callback         = None
         
         self.deferred_stamp_generating = False
@@ -668,21 +671,29 @@ class LXMessage:
 
         return msgpack.packb(container)
 
-
     def write_to_directory(self, directory_path):
         file_name = RNS.hexrep(self.hash, delimit=False)
         file_path = directory_path+"/"+file_name
+        tmp_path  = file_path+".tmp."+str(os.getpid() or time.time())+"."+RNS.hexrep(os.urandom(8), delimit=False)
 
-        try:
-            file = open(file_path, "wb")
-            file.write(self.packed_container())
-            file.close()
+        with self.__persist_lock:
+            try:
+                with open(tmp_path, "wb") as file:
+                    file.write(self.packed_container())
+                    file.flush()
+                    try: os.fsync(file.fileno())
+                    except OSError as e: RNS.log(f"Error while waiting for persist fsync for {self}: {e}", RNS.LOG_WARNING)
 
-            return file_path
+                os.replace(tmp_path, file_path)
+                return file_path
 
-        except Exception as e:
-            RNS.log("Error while writing LXMF message to file \""+str(file_path)+"\". The contained exception was: "+str(e), RNS.LOG_ERROR)
-            return None
+            except Exception as e:
+                try:
+                    if os.path.exists(tmp_path): os.unlink(tmp_path)
+                except Exception as e: RNS.log(f"Error while cleaning temporary file {tmp_path} for {self}: {e}", RNS.LOG_ERROR)
+
+                RNS.log(f"Error while writing LXMF message to file \"{file_path}\". The contained exception was: {e}", RNS.LOG_ERROR)
+                return None
 
     def as_uri(self, finalise=True):
         if not self.packed:
@@ -789,6 +800,10 @@ class LXMessage:
         message.set_title_from_bytes(title_bytes)
         message.set_content_from_bytes(content_bytes)
 
+        try:
+            if source_identity != None: message.source_blackholed = RNS.Reticulum.get_instance().is_blackholed(source_identity)
+        except Exception as e: RNS.log(f"Could not determine message source blackhole status: {e}", RNS.LOG_WARNING)
+
         try:
             if source:
                 if source.identity.validate(signature, signed_part):

LXMF/Utilities/lxmd.py

@@ -87,12 +87,17 @@ def apply_config():
         else:
             active_configuration["peer_announce_interval"] = None
         
+        if "lxmf" in lxmd_config and "stamp_cost" in lxmd_config["lxmf"]:
+            active_configuration["peer_stamp_cost"] = max(1, lxmd_config["lxmf"].as_int("stamp_cost"))
+        else:
+            active_configuration["peer_stamp_cost"] = 12
+        
         if "lxmf" in lxmd_config and "delivery_transfer_max_accepted_size" in lxmd_config["lxmf"]:
             active_configuration["delivery_transfer_max_accepted_size"] = lxmd_config["lxmf"].as_float("delivery_transfer_max_accepted_size")
             if active_configuration["delivery_transfer_max_accepted_size"] < 0.38:
                 active_configuration["delivery_transfer_max_accepted_size"] = 0.38
         else:
-            active_configuration["delivery_transfer_max_accepted_size"] = 1000
+            active_configuration["delivery_transfer_max_accepted_size"] = 1
 
         if "lxmf" in lxmd_config and "on_inbound" in lxmd_config["lxmf"]:
             active_configuration["on_inbound"] = lxmd_config["lxmf"]["on_inbound"]
@@ -130,6 +135,21 @@ def apply_config():
         else:
             active_configuration["autopeer_maxdepth"] = None
 
+        if "propagation" in lxmd_config and "sequential_pn_stamp_validation" in lxmd_config["propagation"]:
+            active_configuration["sequential_pn_stamp_validation"] = lxmd_config["propagation"].as_bool("sequential_pn_stamp_validation")
+        else:
+            active_configuration["sequential_pn_stamp_validation"] = True
+
+        if "propagation" in lxmd_config and "static_peers_bypass_sequential" in lxmd_config["propagation"]:
+            active_configuration["static_peers_bypass_sequential"] = lxmd_config["propagation"].as_bool("static_peers_bypass_sequential")
+        else:
+            active_configuration["static_peers_bypass_sequential"] = True
+
+        if "propagation" in lxmd_config and "max_inbound_syncs" in lxmd_config["propagation"]:
+            active_configuration["max_inbound_syncs"] = max(1, lxmd_config["propagation"].as_int("max_inbound_syncs"))
+        else:
+            active_configuration["max_inbound_syncs"] = 3
+
         if "propagation" in lxmd_config and "announce_interval" in lxmd_config["propagation"]:
             active_configuration["node_announce_interval"] = lxmd_config["propagation"].as_int("announce_interval")*60
         else:
@@ -285,8 +305,7 @@ def lxmf_delivery(lxm):
             processing_command = command+" \""+written_path+"\""
             return_code = subprocess.call(shlex.split(processing_command), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
 
-        else:
-            RNS.log("No action defined for inbound messages, ignoring", RNS.LOG_DEBUG)
+        else: RNS.log("No action defined for inbound messages, ignoring", RNS.LOG_DEBUG)
 
     except Exception as e:
         RNS.log("Error occurred while processing received message "+str(lxm)+". The contained exception was: "+str(e), RNS.LOG_ERROR)
@@ -307,7 +326,7 @@ def program_setup(configdir = None, rnsconfigdir = None, run_pn = False, on_inbo
     if configdir == None:
         if os.path.isdir("/etc/lxmd") and os.path.isfile("/etc/lxmd/config"):
             configdir = "/etc/lxmd"
-        elif os.path.isdir(RNS.Reticulum.userdir+"/.config/lxmd") and os.path.isfile(Reticulum.userdir+"/.config/lxmd/config"):
+        elif os.path.isdir(RNS.Reticulum.userdir+"/.config/lxmd") and os.path.isfile(RNS.Reticulum.userdir+"/.config/lxmd/config"):
             configdir = RNS.Reticulum.userdir+"/.config/lxmd"
         else:
             configdir = RNS.Reticulum.userdir+"/.lxmd"
@@ -383,6 +402,9 @@ def program_setup(configdir = None, rnsconfigdir = None, run_pn = False, on_inbo
         propagation_limit = active_configuration["propagation_transfer_max_accepted_size"],
         propagation_cost = active_configuration["propagation_stamp_cost_target"],
         propagation_cost_flexibility = active_configuration["propagation_stamp_cost_flexibility"],
+        sequential_validation = active_configuration["sequential_pn_stamp_validation"],
+        static_sequential = not active_configuration["static_peers_bypass_sequential"],
+        max_inbound_syncs = active_configuration["max_inbound_syncs"],
         peering_cost = active_configuration["peering_cost"],
         max_peering_cost = active_configuration["remote_peering_cost_max"],
         sync_limit = active_configuration["propagation_sync_max_accepted_size"],
@@ -397,7 +419,8 @@ def program_setup(configdir = None, rnsconfigdir = None, run_pn = False, on_inbo
     for destination_hash in active_configuration["ignored_lxmf_destinations"]:
         message_router.ignore_destination(destination_hash)
 
-    lxmf_destination = message_router.register_delivery_identity(identity, display_name=active_configuration["display_name"])
+    lxmf_destination = message_router.register_delivery_identity(identity, display_name=active_configuration["display_name"],
+                                                                 stamp_cost=active_configuration["peer_stamp_cost"])
 
     RNS.Identity.remember(
         packet_hash=None,
@@ -770,7 +793,7 @@ def get_status(remote=None, configdir=None, rnsconfigdir=None, verbosity=0, quie
                 srxb = RNS.prettysize(p["rx_bytes"]); stxb = RNS.prettysize(p["tx_bytes"]); pmo = pm["offered"]; pmout = pm["outgoing"]
                 pmi  = pm["incoming"]; pmuh = pm["unhandled"]; ar = round(p["acceptance_rate"]*100, 2)
                 if p["name"] == None: nn = ""
-                else: nn = p["name"].strip().replace("\n", "").replace("\r", "")
+                else: nn = sanitize_name(p["name"])
                 if len(nn) > 45: nn = f"{nn[:45]}..."
                 print(f"{ind}{t}{RNS.prettyhexrep(peer_id)}")
                 if len(nn): print(f"{ind*2}Name       : {nn}")
@@ -822,7 +845,7 @@ def _remote_init(configdir=None, rnsconfigdir=None, verbosity=0, quietness=0, id
     if identity_path == None:
         if configdir == None:
             if os.path.isdir("/etc/lxmd") and os.path.isfile("/etc/lxmd/config"): configdir = "/etc/lxmd"
-            elif os.path.isdir(RNS.Reticulum.userdir+"/.config/lxmd") and os.path.isfile(Reticulum.userdir+"/.config/lxmd/config"): configdir = RNS.Reticulum.userdir+"/.config/lxmd"
+            elif os.path.isdir(RNS.Reticulum.userdir+"/.config/lxmd") and os.path.isfile(RNS.Reticulum.userdir+"/.config/lxmd/config"): configdir = RNS.Reticulum.userdir+"/.config/lxmd"
             else: configdir = RNS.Reticulum.userdir+"/.lxmd"
 
         configpath   = configdir+"/config"
@@ -1060,6 +1083,33 @@ autopeer_maxdepth = 6
 
 # from_static_only = True
 
+# By default, stamp validation jobs for PN
+# sync batches will run sequentially. If
+# a peer offers messages while another batch
+# is already processing, it will receive a
+# throttle response to indicate that it can
+# retry later. If you have a fast system, you
+# can disable this to accept and validate
+# everything as soon as it is offered.
+
+# sequential_pn_stamp_validation = yes
+
+# You can configure whether static peers are
+# allowed have their PN syncs processed as
+# soon as they are offered, regardless of
+# whether another batch is already validating.
+
+# static_peers_bypass_sequential = yes
+
+# You can configure how many concurrent inbound
+# propagation sync transfers will be accepted.
+# Once this number is reached, nodes offering
+# messages will receive a throttle response to
+# indicate that they can retry later. On a slow
+# system, it's a good idea to change this to 1.
+
+# max_inbound_syncs = 3
+
 # By default, any destination is allowed to
 # connect and download messages, but you can
 # optionally restrict this. If you enable
@@ -1089,13 +1139,18 @@ announce_at_start = no
 
 # announce_interval = 360
 
+# You can configure the required stamp cost for
+# incoming messages.
+
+# stamp_cost = 12
+
 # The maximum accepted unpacked size for mes-
 # sages received directly from other peers,
 # specified in kilobytes. Messages larger than
 # this will be rejected before the transfer
 # begins.
 
-delivery_transfer_max_accepted_size = 1000
+delivery_transfer_max_accepted_size = 1
 
 # You can configure an external program to be run
 # every time a message is received. The program
@@ -1122,5 +1177,77 @@ loglevel = 4
 
 """
 
-if __name__ == "__main__":
-    main()
+import re
+import unicodedata
+STRIP_BLOCKS_RE = re.compile(
+    '['
+    '\U0001F600-\U0001F64F'  # Emoticons
+    '\U0001F300-\U0001F5FF'  # Misc Symbols & Pictographs
+    '\U0001F680-\U0001F6FF'  # Transport & Map Symbols
+    '\U0001F700-\U0001F77F'  # Alchemical Symbols
+    '\U0001F780-\U0001F7FF'  # Geometric Shapes Extended
+    '\U0001F800-\U0001F8FF'  # Supplemental Arrows-C
+    '\U0001F900-\U0001F9FF'  # Supplemental Symbols & Pictographs
+    '\U0001FA00-\U0001FA6F'  # Chess Symbols
+    '\U0001FA70-\U0001FAFF'  # Symbols & Pictographs Extended-A
+    '\U0001F1E0-\U0001F1FF'  # Flags (iOS/regional indicators)
+    '\u2600-\u26FF'          # Misc Symbols (☀, ☁, ☂, etc.)
+    '\u2700-\u27BF'          # Dingbats (✂, ✈, ✉, etc.)
+    '\uFE00-\uFE0F'          # Variation Selectors
+    '\U000E0100-\U000E01EF'  # Variation Selectors Supplement
+    '\U0001F3FB-\U0001F3FF'  # Emoji modifiers (skin tones)
+    ']+',
+    flags=re.UNICODE
+)
+
+STRIP_CONTROL_RE = re.compile(
+    '['
+    '\x00-\x08'      # C0 controls (NUL-BS)
+    '\x0B\x0C'       # VT, FF
+    '\x0E-\x1F'      # C0 controls (SO-US)
+    '\x7F-\x9F'      # DEL and C1 controls
+    '\u200B-\u200F'  # Zero-width chars, LRM, RLM, etc.
+    '\u202A-\u202E'  # Bidi embedding controls
+    '\u2060-\u206F'  # Format chars (word joiner, etc.)
+    '\uFEFF'         # BOM / Zero Width NBSP
+    '\uFFF0-\uFFF8'  # Specials
+    ']+',
+    flags=re.UNICODE
+)
+
+STRIP_PRIVATE_RE = re.compile(
+    '['
+    '\uD800-\uDFFF'         # Surrogates
+    '\uE000-\uF8FF'         # Private Use Area
+    '\uF900-\uFAFF'         # CJK Compatibility Ideographs (keep? strip for safety)
+    '\uFE10-\uFE1F'         # Vertical Forms
+    '\uFE20-\uFE2F'         # Combining Half Marks
+    '\U000F0000-\U000FFFFF' # Supplementary Private Use Area-A
+    '\U00100000-\U0010FFFF' # Supplementary Private Use Area-B
+    ']+',
+    flags=re.UNICODE
+)
+
+def sanitize_name(name):
+    if name is None: return None
+    name = str(name)
+    name = unicodedata.normalize('NFKC', name)    
+    result = []
+    for char in name:
+        cat = unicodedata.category(char)
+        cat_prefix = cat[0] if cat else 'C'
+        if cat_prefix in ('L', 'N', 'P'): result.append(char)
+        elif cat == 'Zs': result.append(' ')
+        elif cat in ('Zl', 'Zp'): result.append(' ')
+        elif cat == 'Mc': result.append(char)
+        elif cat == 'Lm': result.append(char)
+
+    name = ''.join(result)    
+    name = STRIP_BLOCKS_RE.sub('', name)
+    name = STRIP_CONTROL_RE.sub('', name)
+    name = STRIP_PRIVATE_RE.sub('', name)
+    name = re.sub(r'\s+', ' ', name)
+    name = name.strip()
+    return name
+
+if __name__ == "__main__": main()

LXMF/_version.py

@@ -1 +1 @@
-__version__ = "0.9.8"
+__version__ = "1.1.0"

Makefile

@@ -23,8 +23,13 @@ build_sdist:
 
 build_spkg: remove_symlinks build_sdist create_symlinks
 
-release: remove_symlinks build_wheel create_symlinks
+release: remove_symlinks build_wheel build_spkg create_symlinks
 
 upload:
+	@echo Ready to publish release over Reticulum
+	@read VOID
+	rngit release rns://7649a50d84610232d1416b41d2896aff/reticulum/lxmf create $$(python setup.py --getversion):dist --name lxmf
+
+upload-pip:
 	@echo Uploading to PyPi...
-	twine upload dist/*
+	twine upload dist/*.whl dist/*.tar.gz

setup.py

@@ -1,3 +1,4 @@
+import sys
 import setuptools
 
 with open("README.md", "r") as fh:
@@ -5,6 +6,10 @@ with open("README.md", "r") as fh:
 
 exec(open("LXMF/_version.py", "r").read())
 
+if "--getversion" in sys.argv:
+    print(__version__, end="")
+    exit(0)
+
 setuptools.setup(
     name="lxmf",
     version=__version__,
@@ -26,6 +31,6 @@ setuptools.setup(
             'lxmd=LXMF.Utilities.lxmd:main',
         ]
     },
-    install_requires=["rns>=1.2.5"],
+    install_requires=["rns>=1.4.0"],
     python_requires=">=3.7",
 )