Implement SYNC mechanism

README.md

@@ -3,10 +3,8 @@
 A system for finding application peers and resolving names over mesh-like networks
 
 TODO:
-- Create a NameInfoStore, along with a FilesystemNameInfoStore and maybe a memory one too
-- Add `SYNC` request with `DOMAINS` response for quickly getting a new server peer up to date on domain information
 - Add propogation rate limit
 - Verify CertifiedNameInfo against system trust
 - Server class should implement (or inherit) a "Service" class, with another implementation being "DaemonClient" when a Riddle Daemon is implemented
 - Build tooling for generating and propogating NameInfo.
-- OpenPGP domains, in the form `[fingerprint].rns` i.e. "c3a6-5e46-7b54-77df-3c4c-9790-4d22-b3ca-5b32-ff66.rns"
+- OpenPGP domains, in the form `[fingerprint].rns` i.e. "c3a6-5e46-7b54-77df-3c4c-9790-4d22-b3ca-5b32-ff66.rns"

src/lib/Client.vala

@@ -9,8 +9,6 @@ namespace Riddle {
         private Invercargill.Sequence<Message> join_messages = new Invercargill.Sequence<Message>();
         private InetSocketAddress? self_server = null;
 
-        
-
         public Client() {
             discoverers.add (new KnownHostDiscoverer());
             discoverers.add (new LanDiscoverer());
@@ -55,7 +53,6 @@ namespace Riddle {
             foreach (var message in j_msgs) {
                 raw_request(message, single(address), 5000, MessageType.OK);
             }
-
         }
 
         public int join(string group, uint port) {
@@ -134,6 +131,38 @@ namespace Riddle {
 
         }
 
+        public void sync(InetSocketAddress server, NameInfoStore store) throws Error {
+            var socket = new SocketClient().connect(server);
+            var dis = new DataInputStream(socket.input_stream);
+            var dos = new DataOutputStream(socket.output_stream);
+
+            var existing = store.get_all().select<string>(i => @"$(i.name) $(i.effective.format_iso8601())");
+            var request = new Message(MessageType.SYNC, new string[0], existing.to_array());
+
+            request.send(dos);
+            var reply = Message.from_stream(dis);
+
+            if(reply.message_type != MessageType.DOMAINS) {
+                warning(@"Bad response from $(server.address) for SYNC request.");
+            }
+
+            foreach (var item in reply.items) {
+                var parts = item.split(" ", 2);
+                var name = parts[0];
+                var data = parts[1];
+                
+                NameInfo info;
+                if(name.has_suffix(".rns")) {
+                    info = new DecentralisedNameInfo.from_string(data);
+                }
+                else {
+                    info = new CertifiedNameInfo.from_string(data);
+                }
+
+                store.save_name(info);
+            }
+        }
+
         internal static bool server_equals(InetSocketAddress a1, InetSocketAddress a2) {
             return (a1.address.equal(a2.address) && a1.port == a2.port);
         }

src/lib/DecentralisedNameInfo.vala

@@ -36,6 +36,20 @@ namespace Riddle {
 
         }
 
+        public DecentralisedNameInfo.from_trusted_string(string encoded_name_info) throws NameInfoError {
+            raw_data = Base64.decode (encoded_name_info);
+            public_key = raw_data[0:Sodium.Asymmetric.Signing.PUBLIC_KEY_BYTES];
+            var signed_message = raw_data[Sodium.Asymmetric.Signing.PUBLIC_KEY_BYTES:];
+            var data = Sodium.Asymmetric.Signing.verify (signed_message, public_key);
+            if(data == null) {
+                throw new NameInfoError.BAD_DATA("Invalid signature");
+            }
+            parse_base_info ((string)data);
+            if(!name.has_suffix (".rns")) {
+                throw new NameInfoError.INVALID("Decentralised names must end in '.rns'.");
+            }
+        }
+
         public DecentralisedNameInfo(string name, string key, DateTime expiry, Enumerable<NameProperty> properties) {
             var keys = key.split(":");
             public_key = Base64.decode(keys[0]);

src/lib/FilesystemNameInfoStore.vala

@@ -1,3 +1,4 @@
+using Invercargill;
 
 namespace Riddle {
 
@@ -19,13 +20,7 @@ namespace Riddle {
                 var name_stream = new DataInputStream (file.read ());
                 var data = name_stream.read_line();
                 name_stream.close();
-
-                if(name.has_suffix (".rns")) {
-                    info = new DecentralisedNameInfo.from_string (data);
-                }
-                else {
-                    info = new CertifiedNameInfo.from_string (data);
-                }
+                info = build_info (name, data);
             }
             catch(Error e) {
                 warning(@"Error while reading name from filesystem store: $(e.message)");
@@ -45,11 +40,48 @@ namespace Riddle {
                 warning(@"Error while writing name to filesystem store: $(e.message)");
             }
         }
+        public override Enumerable<NameInfo> get_all() {
+            try {
+                var dir = Dir.open (path);
+                return new Generator<NameInfo>(() => {
+                    var file_name = dir.read_name ();
+                    if(file_name == null) {
+                        return GeneratorResult<NameInfo>.end();
+                    }
+                    return GeneratorResult.skip_if_null(get_name(file_name.replace (".nameinfo", "")));
+                });
+            }
+            catch(Error e) {
+                warning(@"Error while reading filesystem store: $(e.message)");
+                return Invercargill.empty<NameInfo> ();
+            }
+        }
+        public override void clean() {
+            var now = new DateTime.now_utc();
+            var names = get_all().to_sequence();
+            foreach (var name in names) {
+                if(name.expires.difference(now) < 0) {
+                    try {
+                        get_name_file(name.name).delete();
+                    }
+                    catch(Error e) {
+                        warning(@"Error while deleting name from store: $(e.message)");
+                    }
+                }
+            }
+        }
 
         private File get_name_file(string name) {
             return File.new_for_path (@"$path/$name.nameinfo");
         }
 
+        private NameInfo build_info(string name, string data) throws NameInfoError {
+            if(name.has_suffix (".rns")) {
+                return new DecentralisedNameInfo.from_trusted_string (data);
+            }
+            return new CertifiedNameInfo.from_string (data);
+        }
+
     }
 
 }

src/lib/Message.vala

@@ -11,12 +11,14 @@ namespace Riddle {
         WHO_IS,
         RIDDLE,
         CALLBACK,
+        SYNC,
 
         OK,
         ERROR,
         SEE_ALSO,
         ANSWER,
         SOLVED,
+        DOMAINS,
         NOT_ACCEPTED,
         UNKNOWN;
 
@@ -36,6 +38,8 @@ namespace Riddle {
                     return MessageType.RIDDLE;
                 case "CALLBACK":
                     return MessageType.CALLBACK;
+                case "SYNC":
+                    return MessageType.SYNC;
                 case "OK":
                     return MessageType.OK;
                 case "ERROR":
@@ -46,6 +50,8 @@ namespace Riddle {
                     return MessageType.ANSWER;
                 case "SOLVED":
                     return MessageType.SOLVED;
+                case "DOMAINS":
+                    return MessageType.DOMAINS;
                 case "NOT-ACCEPTED":
                     return MessageType.NOT_ACCEPTED;
                 case "UNKNOWN":
@@ -71,6 +77,8 @@ namespace Riddle {
                     return "RIDDLE";
                 case MessageType.CALLBACK:
                     return "CALLBACK";
+                case MessageType.SYNC:
+                    return "SYNC";
                 case MessageType.OK:
                     return  "OK";
                 case MessageType.ERROR:
@@ -81,6 +89,8 @@ namespace Riddle {
                     return "ANSWER";
                 case MessageType.SOLVED:
                     return "SOLVED";
+                case MessageType.DOMAINS:
+                    return "DOMAINS";
                 case MessageType.NOT_ACCEPTED:
                     return "NOT-ACCEPTED";
                 case MessageType.UNKNOWN:

src/lib/NameInfoStore.vala

@@ -1,3 +1,5 @@
+using Invercargill;
+
 
 namespace Riddle {
 
@@ -6,6 +8,8 @@ namespace Riddle {
         public abstract bool has_name(string name);
         public abstract NameInfo? get_name(string name);
         public abstract void save_name(NameInfo name_info);
+        public abstract Enumerable<NameInfo> get_all();
+        public abstract void clean();
 
         public virtual bool is_outdated(string name, DateTime effective) {
             if(!has_name(name)) {

src/lib/Server.vala

@@ -6,6 +6,7 @@ namespace Riddle {
 
         public const int REGISTRATION_TIMEOUT_US = 600000000;
         public const int GOSSIP_INTERVAL_US = 2000000;
+        public const int SYNC_INTERVAL_US = 2000000;
         public const string RIDDLE_SERVER_GROUP = "Riddle";
 
         private SocketService service;
@@ -15,12 +16,13 @@ namespace Riddle {
         private Client client;
 
         private Fifo<Gossip> gossip = new Fifo<Gossip>();
+        private Fifo<InetSocketAddress> sync_queue = new Fifo<InetSocketAddress>();
         private Gee.HashSet<string> gossip_idempotency_tokens = new Gee.HashSet<string>();
-        private Thread<bool> gossip_thread;
         private uint16 port;
         
         
         public Server(string address, uint16 port, NameInfoStore name_store) throws Error {
+            name_store.clean();
             this.names = name_store;
             this.port = port;
             var add = new InetAddress.from_string(address);
@@ -29,7 +31,8 @@ namespace Riddle {
 
             client = new Client.with_server(new InetSocketAddress(add, port));
             client.join(RIDDLE_SERVER_GROUP, port);
-            gossip_thread = new Thread<bool>("Gossip", spread_gossip);
+            new Thread<bool>("Gossip", spread_gossip);
+            new Thread<bool>("NameSync", server_sync);
         }
 
         public async void start() {
@@ -77,6 +80,21 @@ namespace Riddle {
             return true;
         }
 
+        private bool server_sync() {
+            foreach (var server in sync_queue) {
+                try {
+                    lock(names) {
+                        client.sync(server, names);
+                    }
+                }
+                catch(Error e) {
+                    warning(@"Error while attempting to auto-sync with server: $(e.message)");
+                }
+                Thread.usleep(SYNC_INTERVAL_US);
+            }
+            return true;
+        }
+
         private Message service_message(Message msg, InetSocketAddress origin) throws Error {
 
             switch (msg.message_type) {
@@ -94,6 +112,8 @@ namespace Riddle {
                     return handle_riddle(msg, origin);
                 case MessageType.CALLBACK:
                     return handle_callback(msg);
+                case MessageType.SYNC:
+                    return handle_sync(msg);
                 default:
                     return new Message(MessageType.ERROR, new string[] { "unknown-command" }, new string[0]);
             }
@@ -221,6 +241,19 @@ namespace Riddle {
             return new Message(MessageType.ERROR, new string[] { "not-implemented" }, new string[0]);
         }
 
+        private Message handle_sync(Message msg) throws Error {
+            var query = names.get_all();
+            foreach (var existing in msg.items) {
+                var parts = existing.split(" ", 2);
+                var name = parts[0];
+                var effective_date = new DateTime.from_iso8601(parts[1], null);
+
+                query = query.where(info => info.name != name || info.effective.difference(effective_date) > 0);
+            }
+
+            return new Message(MessageType.DOMAINS, new string[0], query.select<string>(i => @"$(i.name) $(i.get_encoded())").to_array());
+        }
+
         private void cleanup_registrations() {
             lock(registrations) {
                 var copy = new Gee.HashMap<string, Invercargill.Sequence<Registration>>();
@@ -245,6 +278,10 @@ namespace Riddle {
 
                 registrations[group].add(reg);
             }
+
+            if(group == RIDDLE_SERVER_GROUP) {
+                sync_queue.push(address);
+            }
         }
 
         private void remove_registration(string group, InetSocketAddress address) {