Add sync() function to enable consistent cross-client views

README.md

@@ -413,6 +413,46 @@ zookeeper.setData('/test/demo', null, 2, function (error, stat) {
 
 ---
 
+#### void sync(path, callback)
+
+Synchronize the client's view of the given path with the ZooKeeper leader.
+
+It is possible for a client to read a stale value for a given path, even
+after another client has received confirmation that the new value has been
+committed. Only a quorum of servers has to agree to a given change for a
+write to be confirmed, and we might be connected to a server that hasn't
+seen the update yet.
+
+The callback function will be called once we confirm that any read of the
+given path will return a value at least as new as when sync() was called.
+
+**Arguments**
+
+* path `String` - Path of the node.
+* callback(error) `Function` - The callback function.
+
+**Example**
+
+```javascript
+zookeeper.sync('/test/demo', function (error) {
+    if (error) {
+        console.log(error.stack);
+        return;
+    }
+
+    zookeeper.getData('/test/demo', function (error, data, stat) {
+        if (error) {
+            console.log(error.stack);
+            return;
+        }
+
+        console.log('Fresh value read.');
+    });
+});
+```
+
+---
+
 #### void getACL(path, callback)
 
 Retrieve the list of [ACL](#acl) and stat of the node of the given path.

index.js

@@ -422,6 +422,42 @@ Client.prototype.create = function (path, data, acls, mode, callback) {
     );
 };
 
+/**
+ * Synchronize this client's view of the given path with the
+ * ZooKeeper leader. From the callback, any read will see
+ * data at least as new as when this function was called.
+ *
+ * @method sync
+ * @param path {String} The node path.
+ * @param callback {Function} The callback function.
+ */
+Client.prototype.sync = function (path, callback) {
+    Path.validate(path);
+
+    assert(typeof callback === 'function', 'callback must be a function.');
+
+    var self = this,
+        header = new jute.protocol.RequestHeader(),
+        payload = new jute.protocol.SyncRequest(),
+        request;
+
+    header.type = jute.OP_CODES.SYNC;
+
+    payload.path = path;
+
+    request = new jute.Request(header, payload);
+
+    attempt(
+        self,
+        function (attempts, next) {
+            self.connectionManager.queue(request, function (error, response) {
+                next(error);
+            });
+        },
+        callback
+    );
+};
+
 /**
  * Delete a node with the given path. If version is not -1, the request will
  * fail when the provided version does not match the server version.

lib/ConnectionManager.js

@@ -528,6 +528,9 @@ ConnectionManager.prototype.onSocketData = function (buffer) {
                 case jute.OP_CODES.GET_DATA:
                     responsePayload = new jute.protocol.GetDataResponse();
                     break;
+                case jute.OP_CODES.SYNC:
+                    responsePayload = new jute.protocol.SyncResponse();
+                    break;
                 case jute.OP_CODES.SET_ACL:
                     responsePayload = new jute.protocol.SetACLResponse();
                     break;

lib/jute/specification.json

@@ -94,6 +94,12 @@
             { "name" : "data", "type" : "buffer" },
             { "name" : "stat", "type" : "data.Stat" }
         ],
+        "SyncRequest" : [
+            { "name" : "path", "type" : "ustring" }
+        ],
+        "SyncResponse" : [
+            { "name" : "path", "type" : "ustring" }
+        ],
         "GetACLRequest" : [
             { "name" : "path", "type" : "ustring" }
         ],