Updated versions

LXMF/Handlers.py

@@ -18,7 +18,7 @@ class LXMFDeliveryAnnounceHandler:
             self.lxmrouter.update_stamp_cost(destination_hash, stamp_cost)
 
         except Exception as e:
-            RNS.log(f"An error occurred while trying to decode announced stamp cost. The contained exception was: {e}", RNS.LOG_ERROR)
+            RNS.log(f"Could not decode stamp cost from announce data. The contained exception was: {e}", RNS.LOG_DEBUG)
 
         for lxmessage in self.lxmrouter.pending_outbound:
             if destination_hash     == lxmessage.destination_hash:
@@ -41,6 +41,18 @@ class LXMFPropagationAnnounceHandler:
     def received_announce(self, destination_hash, announced_identity, app_data, announce_packet_hash, is_path_response):
         try:
             if type(app_data) == bytes:
+                if destination_hash == self.lxmrouter.get_outbound_propagation_node():
+                    if pn_announce_data_is_valid(app_data):
+                        for lxmessage in self.lxmrouter.pending_outbound:
+                            if lxmessage.method == LXMessage.PROPAGATED:
+                                lxmessage.next_delivery_attempt = time.time()
+
+                                def outbound_trigger():
+                                    while self.lxmrouter.outbound_processing_lock.locked(): time.sleep(0.1)
+                                    self.lxmrouter.process_outbound()
+
+                                threading.Thread(target=outbound_trigger, daemon=True).start()
+
                 if self.lxmrouter.propagation_node:
                     if pn_announce_data_is_valid(app_data):
                         data                               = msgpack.unpackb(app_data)

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
@@ -128,15 +162,14 @@ def display_name_from_app_data(app_data=None):
                     if dn == None: return None
                     else:
                         try:
-                            decoded = dn.decode("utf-8")
+                            decoded = dn.decode("utf-8").replace("\x00", "").strip()
                             return decoded
                         except Exception as e:
                             RNS.log(f"Could not decode display name in included announce data. The contained exception was: {e}", RNS.LOG_ERROR)
                             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,18 +94,18 @@ 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))
 
         self.pending_inbound       = []
         self.pending_outbound      = []
-        self.failed_outbound       = []
         self.direct_links          = {}
         self.backchannel_links     = {}
         self.delivery_destinations = {}
@@ -132,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
@@ -149,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 = {}
@@ -161,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
@@ -238,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 = {}
@@ -487,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:
@@ -853,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
@@ -871,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()
 
@@ -891,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)
@@ -911,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 +973,17 @@ class LXMRouter:
             for link in inactive_links:
                 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)
@@ -957,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()
@@ -997,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)
 
@@ -1178,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)
@@ -1193,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)
@@ -1204,18 +1256,18 @@ 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:
@@ -1235,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)
@@ -1279,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)
@@ -1338,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()
@@ -1354,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)
 
@@ -1591,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)
@@ -1600,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:
-            return False
+        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:
@@ -1645,6 +1746,10 @@ class LXMRouter:
     def handle_outbound(self, lxmessage):
         destination_hash = lxmessage.get_destination().hash
 
+        if lxmessage.desired_method == LXMessage.PROPAGATED and not self.outbound_propagation_node:
+            self.fail_message(lxmessage)
+            raise IOError("Attempt to send propagated message with no outbound propagation node configured")
+
         if lxmessage.stamp_cost == None:
             if destination_hash in self.outbound_stamp_costs:
                 stamp_cost = self.outbound_stamp_costs[destination_hash][1]
@@ -1732,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]
@@ -1800,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)
@@ -1851,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()
@@ -1875,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)
 
@@ -1983,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))
@@ -2081,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()
@@ -2104,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):
@@ -2140,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:
@@ -2182,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)
@@ -2245,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]
@@ -2289,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])
 
@@ -2319,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]
@@ -2329,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]
@@ -2393,22 +2565,15 @@ class LXMRouter:
         RNS.log(str(lxmessage)+" failed to send", RNS.LOG_DEBUG)
 
         lxmessage.progress = 0.0
-        if lxmessage in self.pending_outbound:
-            self.pending_outbound.remove(lxmessage)
-
-        self.failed_outbound.append(lxmessage)
-
-        if lxmessage.state != LXMessage.REJECTED:
-            lxmessage.state = LXMessage.FAILED
-
+        if lxmessage in self.pending_outbound: self.pending_outbound.remove(lxmessage)
+        if lxmessage.state != LXMessage.REJECTED: lxmessage.state = LXMessage.FAILED
         if lxmessage.failed_callback != None and callable(lxmessage.failed_callback):
             lxmessage.failed_callback(lxmessage)
 
     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:
@@ -2417,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/LXStamper.py

@@ -2,9 +2,11 @@ import RNS
 import RNS.vendor.umsgpack as msgpack
 
 import os
+import sys
 import time
 import math
 import itertools
+import contextlib
 import multiprocessing
 
 WORKBLOCK_EXPAND_ROUNDS         = 3000
@@ -12,9 +14,38 @@ WORKBLOCK_EXPAND_ROUNDS_PN      = 1000
 WORKBLOCK_EXPAND_ROUNDS_PEERING = 25
 STAMP_SIZE                      = RNS.Identity.HASHLENGTH//8
 PN_VALIDATION_POOL_MIN_SIZE     = 256
+USE_WORKER_MANAGER              = False
 
 active_jobs = {}
 
+if sys.version_info[0] >= 3 and sys.version_info[1] >= 14:
+    USE_WORKER_MANAGER          = True
+
+@contextlib.contextmanager
+def worker_context(ctx, stamp_cost, workblock, message_id):
+    stop_event = ctx.Event()
+    result_queue = ctx.Queue(1)
+    rounds_queue = ctx.Queue()
+
+    active_jobs[message_id] = [stop_event, result_queue]
+    job_procs = []
+
+    try: yield stop_event, result_queue, rounds_queue, job_procs
+    finally:
+        if message_id in active_jobs: del active_jobs[message_id]
+        stop_event.set()
+
+        for p in job_procs:
+            if p.is_alive(): p.terminate()
+        
+        for p in job_procs: p.join(timeout=0.5)
+        
+        result_queue.close()
+        result_queue.join_thread()
+        rounds_queue.close()
+        rounds_queue.join_thread()
+        stop_event.clear()
+
 def stamp_workblock(material, expand_rounds=WORKBLOCK_EXPAND_ROUNDS):
     wb_st = time.time()
     workblock = b""
@@ -100,7 +131,9 @@ def generate_stamp(message_id, stamp_cost, expand_rounds=WORKBLOCK_EXPAND_ROUNDS
 
     if RNS.vendor.platformutils.is_windows() or RNS.vendor.platformutils.is_darwin(): stamp, rounds = job_simple(stamp_cost, workblock, message_id)
     elif RNS.vendor.platformutils.is_android(): stamp, rounds = job_android(stamp_cost, workblock, message_id)
-    else: stamp, rounds = job_linux(stamp_cost, workblock, message_id)
+    else:
+        if USE_WORKER_MANAGER: stamp, rounds = job_linux_managed(stamp_cost, workblock, message_id)
+        else:                  stamp, rounds = job_linux(stamp_cost, workblock, message_id)
     
     duration = time.time() - start_time
     speed = rounds/duration
@@ -176,6 +209,90 @@ def job_simple(stamp_cost, workblock, message_id):
     
     return pstamp, rounds
 
+def job_linux_managed(stamp_cost, workblock, message_id):
+    ctx = multiprocessing.get_context("fork")
+    cores = multiprocessing.cpu_count()
+    jobs = cores if cores <= 12 else int(cores/2)
+    
+    allow_kill = True
+    stamp = None
+    total_rounds = 0
+    
+    with worker_context(ctx, stamp_cost, workblock, message_id) as (stop_event, result_queue, rounds_queue, job_procs):
+        RNS.log(f"Starting {jobs} stamp generation workers", RNS.LOG_DEBUG)
+        
+        def job(stop_event, sc, wb, worker_id):
+            terminated = False
+            rounds = 0
+            pstamp = os.urandom(256//8)
+
+            def sv(s, c, w):
+                target = 0b1<<256-c; m = w+s
+                result = RNS.Identity.full_hash(m)
+                if int.from_bytes(result, byteorder="big") > target: return False
+                else:                                                return True
+
+            while not stop_event.is_set() and not sv(pstamp, sc, wb):
+                pstamp = os.urandom(256//8); rounds += 1
+
+            if not stop_event.is_set():
+                stop_event.set()
+                try: result_queue.put_nowait(pstamp)
+                except: pass
+            
+            try: rounds_queue.put_nowait(rounds)
+            except: pass
+
+        for jpn in range(jobs):
+            p = ctx.Process(target=job, args=(stop_event, stamp_cost, workblock, jpn), daemon=True)
+            job_procs.append(p)
+            p.start()
+        
+        try: stamp = result_queue.get(timeout=None)
+        except Exception as e:
+            RNS.log(f"Failed to get result from workers: {e}", RNS.LOG_ERROR)
+            stamp = None
+        
+        # Collect any potential spurious
+        # results from worker queue.
+        try:
+            while True: result_queue.get_nowait()
+        except: pass
+
+        for j in range(jobs):
+            nrounds = 0
+            try:
+                nrounds = rounds_queue.get(timeout=2)
+            except Exception as e:
+                RNS.log(f"Failed to get round stats part {j}: {e}", RNS.LOG_ERROR)
+            total_rounds += nrounds
+
+        all_exited = False
+        exit_timeout = time.time() + 5
+        while time.time() < exit_timeout:
+            if not any(p.is_alive() for p in job_procs):
+                all_exited = True
+                break
+            time.sleep(0.1)
+
+        if not all_exited:
+            RNS.log("Stamp generation IPC timeout, possible worker deadlock. Terminating remaining processes.", RNS.LOG_ERROR)
+            if allow_kill:
+                for j in range(jobs):
+                    process = job_procs[j]
+                    process.kill()
+            else:
+                return None
+
+        else:
+            for j in range(jobs):
+                process = job_procs[j]
+                process.join()
+                # RNS.log(f"Joined {j} / {process}", RNS.LOG_DEBUG) # TODO: Remove
+
+    return stamp, total_rounds
+
+
 def job_linux(stamp_cost, workblock, message_id):
     allow_kill = True
     stamp = None
@@ -366,30 +483,41 @@ if __name__ == "__main__":
         RNS.log("No cost argument provided", RNS.LOG_ERROR)
         exit(1)
     else:
-        try:
-            cost = int(sys.argv[1])
+        try: cost = int(sys.argv[1])
         except Exception as e:
             RNS.log(f"Invalid cost argument provided: {e}", RNS.LOG_ERROR)
             exit(1)
 
+        rounds = 1
+        if len(sys.argv) > 2:
+            try: rounds = int(sys.argv[2])
+            except Exception as e:
+                RNS.log(f"Invalid rounds argument provided: {e}", RNS.LOG_ERROR)
+                exit(1)
+
     RNS.loglevel = RNS.LOG_DEBUG
-    RNS.log("Testing LXMF stamp generation", RNS.LOG_DEBUG)
-    message_id = os.urandom(32)
-    generate_stamp(message_id, cost)
+    
+    for i in range(rounds):
+        RNS.log("", RNS.LOG_DEBUG)
+        RNS.log(f"Round {i+1} of {rounds}", RNS.LOG_DEBUG)
 
-    RNS.log("", RNS.LOG_DEBUG)
-    RNS.log("Testing propagation stamp generation", RNS.LOG_DEBUG)
-    message_id = os.urandom(32)
-    generate_stamp(message_id, cost, expand_rounds=WORKBLOCK_EXPAND_ROUNDS_PN)
+        RNS.log("Testing LXMF stamp generation", RNS.LOG_DEBUG)
+        message_id = os.urandom(32)
+        generate_stamp(message_id, cost)
 
-    RNS.log("", RNS.LOG_DEBUG)
-    RNS.log("Testing peering key generation", RNS.LOG_DEBUG)
-    message_id = os.urandom(32)
-    generate_stamp(message_id, cost, expand_rounds=WORKBLOCK_EXPAND_ROUNDS_PEERING)
+        RNS.log("", RNS.LOG_DEBUG)
+        RNS.log("Testing propagation stamp generation", RNS.LOG_DEBUG)
+        message_id = os.urandom(32)
+        generate_stamp(message_id, cost, expand_rounds=WORKBLOCK_EXPAND_ROUNDS_PN)
 
-    transient_list = []
-    st = time.time(); count = 10000
-    for i in range(count): transient_list.append(os.urandom(256))
-    validate_pn_stamps(transient_list, 5)
-    dt = time.time()-st; mps = count/dt
-    RNS.log(f"Validated {count} PN stamps in {RNS.prettytime(dt)}, {round(mps,1)} m/s", RNS.LOG_DEBUG)
+        RNS.log("", RNS.LOG_DEBUG)
+        RNS.log("Testing peering key generation", RNS.LOG_DEBUG)
+        message_id = os.urandom(32)
+        generate_stamp(message_id, cost, expand_rounds=WORKBLOCK_EXPAND_ROUNDS_PEERING)
+
+        # transient_list = []
+        # st = time.time(); count = 10000
+        # for i in range(count): transient_list.append(os.urandom(256))
+        # validate_pn_stamps(transient_list, 5)
+        # dt = time.time()-st; mps = count/dt
+        # RNS.log(f"Validated {count} PN stamps in {RNS.prettytime(dt)}, {round(mps,1)} m/s", RNS.LOG_DEBUG)

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.5"
+__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

README.md

@@ -1,8 +1,6 @@
 # Lightweight Extensible Message Format
 
-*This repository is [a public mirror](./MIRROR.md). All development is happening elsewhere.*
-
-LXMF is a simple and flexible messaging format and delivery protocol that allows a wide variety of implementations, while using as little bandwidth as possible. It is built on top of [Reticulum](https://reticulum.network) and offers zero-conf message routing, end-to-end encryption and Forward Secrecy, and can be transported over any kind of medium that Reticulum supports.
+LXMF is a simple and flexible messaging format and delivery protocol that allows a wide variety of implementations, while using as little bandwidth as possible. It is built on top of [Reticulum](a8d24177d946de4f1f0a0fe1af9a1338:/page/repo.mu`g=reticulum|r=reticulum) and offers zero-conf message routing, end-to-end encryption and Forward Secrecy, and can be transported over any kind of medium that Reticulum supports.
 
 LXMF is efficient enough that it can deliver messages over extremely low-bandwidth systems such as packet radio or LoRa. Encrypted LXMF messages can also be encoded as QR-codes or text-based URIs, allowing completely analog *paper message* transport.
 
@@ -87,7 +85,6 @@ lxm_router = LXMF.LXMRouter()
 message = LXMF.LXMessage(destination, source, "This is a short, simple message.")
 
 lxm_router.handle_outbound(message)
-
 ```
 
 The LXM Router then handles the heavy lifting, such as message packing, encryption, delivery confirmation, path lookup, routing, retries and failure notifications.
@@ -125,11 +122,13 @@ You can try out the paper messaging functionality by using the following QR code
 
 The [Sideband](https://unsigned.io/sideband) application allows you to do this easily. After you have imported the identity into an app of your choice, you can scan the following QR code and open it in the app, where it will be decrypted and added as a message.
 
-<p align="center"><img width="50%" src="./docs/paper_msg_test.png"/></p>
+[Paper message QR code](a8d24177d946de4f1f0a0fe1af9a1338:/page/blob.mu`g=reticulum|r=lxmf|ref=HEAD|path=docs/paper_msg_test.png)
 
-You can also find the entire message in <a href="lxm://azNivSwdv4e2aoX3mo2MdTAozuI7BlzrLlHULmnVgpz3dNT9CMPVwgywzCJP8FVogj5j_kU7j7ywuvBNcr45kRTrd19c3iHenmnSDe4VEd6FuGsAiT0Khzl7T81YZHPTDhRNp0FdhDE9AJ7uphw7zKMyqhHHxOxqrYeBeKF66gpPxDceqjsOApvsSwggjcuHBx9OxOBy05XmnJxA1unCKgvNfOFYc1T47luxoY3c0dLOJnJPwZuFRytx2TXlQNZzOJ28yTEygIfkDqEO9mZi5lgev7XZJ0DvgioQxMIyoCm7lBUzfq66zW3SQj6vHHph7bhr36dLOCFgk4fZA6yia2MlTT9KV66Tn2l8mPNDlvuSAJhwDA_xx2PN9zKadCjo9sItkAp8r-Ss1CzoUWZUAyT1oDw7ly6RrzGBG-e3eM3CL6u1juIeFiHby7_3cON-6VTUuk4xR5nwKlFTu5vsYMVXe5H3VahiDSS4Q1aqX7I">this link</a>:
+You can also find the entire message in this link:
 
-`lxm://azNivSwdv4e2aoX3mo2MdTAozuI7BlzrLlHULmnVgpz3dNT9CMPVwgywzCJP8FVogj5j_kU7j7ywuvBNcr45kRTrd19c3iHenmnSDe4VEd6FuGsAiT0Khzl7T81YZHPTDhRNp0FdhDE9AJ7uphw7zKMyqhHHxOxqrYeBeKF66gpPxDceqjsOApvsSwggjcuHBx9OxOBy05XmnJxA1unCKgvNfOFYc1T47luxoY3c0dLOJnJPwZuFRytx2TXlQNZzOJ28yTEygIfkDqEO9mZi5lgev7XZJ0DvgioQxMIyoCm7lBUzfq66zW3SQj6vHHph7bhr36dLOCFgk4fZA6yia2MlTT9KV66Tn2l8mPNDlvuSAJhwDA_xx2PN9zKadCjo9sItkAp8r-Ss1CzoUWZUAyT1oDw7ly6RrzGBG-e3eM3CL6u1juIeFiHby7_3cON-6VTUuk4xR5nwKlFTu5vsYMVXe5H3VahiDSS4Q1aqX7I`
+```
+lxm://azNivSwdv4e2aoX3mo2MdTAozuI7BlzrLlHULmnVgpz3dNT9CMPVwgywzCJP8FVogj5j_kU7j7ywuvBNcr45kRTrd19c3iHenmnSDe4VEd6FuGsAiT0Khzl7T81YZHPTDhRNp0FdhDE9AJ7uphw7zKMyqhHHxOxqrYeBeKF66gpPxDceqjsOApvsSwggjcuHBx9OxOBy05XmnJxA1unCKgvNfOFYc1T47luxoY3c0dLOJnJPwZuFRytx2TXlQNZzOJ28yTEygIfkDqEO9mZi5lgev7XZJ0DvgioQxMIyoCm7lBUzfq66zW3SQj6vHHph7bhr36dLOCFgk4fZA6yia2MlTT9KV66Tn2l8mPNDlvuSAJhwDA_xx2PN9zKadCjo9sItkAp8r-Ss1CzoUWZUAyT1oDw7ly6RrzGBG-e3eM3CL6u1juIeFiHby7_3cON-6VTUuk4xR5nwKlFTu5vsYMVXe5H3VahiDSS4Q1aqX7I
+```
 
 On operating systems that allow for registering custom URI-handlers, you can click the link, and it will be decoded directly in your LXMF client. This works with Sideband on Android.
 
@@ -213,8 +212,6 @@ LXMF is beta software, and should be considered experimental. While it has been
 
 LXMF is actively being developed, and the following improvements and features are currently planned for implementation:
 
-- ~~Update examples in readme to actually work~~
-- ~~Sync affinity based on link speeds and distances, for more intelligently choosing peer sync order~~
 - Sneakernet and physical transport functionality
 - Content Destinations, and easy to use API for group messaging and discussion threads 
 - Write and release full API and protocol documentation

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.1.6"],
+    install_requires=["rns>=1.4.0"],
     python_requires=">=3.7",
 )