Merge lp:~hazmat/txzookeeper/backoff-retry-managed-sanity into lp:~hazmat/txzookeeper/trunk

[Kapil Thangavelu]

Session management and retry improvements.

Lots of improvements and several bug fixes.

Previously managed clientwas returning the underlying session client instead
of retry, so retry wasn't enabled for uses trying to catch/store the client from
connect.

The retry code was using seconds for session timeout instead of milliseconds.

The retry code wasn't taking account of the start of retry properly when
determining max retry.

The retry on persistent error wasn't signaling to the managed client to
establish a new session.

The client code wasn't properly cleaning up txzk handles on conn timeouts.

The session establishment now uses the client session events so even without
activity or active watches the session is restablished.

Session restablishment will now backoff on reconnects for up to 6m.

https://codereview.appspot.com/7307051/

[Kapil Thangavelu (hazmat)]

Reviewers: mp+146938_code.launchpad.net,

Message:
Please take a look.

Description:
Session management and retry improvements.

Lots of improvements and several bug fixes.

Previously managed clientwas returning the underlying session client
instead
of retry, so retry wasn't enabled for uses trying to catch/store the
client from
connect.

The retry code was using seconds for session timeout instead of
milliseconds.

The retry code wasn't taking account of the start of retry properly when

determining max retry.

The retry on persistent error wasn't signaling to the managed client to
establish a new session.

The client code wasn't properly cleaning up txzk handles on conn
timeouts.

The session establishment now uses the client session events so even
without
activity or active watches the session is restablished.

Session restablishment will now backoff on reconnects for up to 6m.

https://code.launchpad.net/~hazmat/txzookeeper/backoff-retry-managed-sanity/+merge/146938

(do not edit description out of merge proposal)

Please review this at https://codereview.appspot.com/7307051/

Affected files:
   A .bzrignore
   A [revision details]
   M debian/changelog
   M setup.py
   M txzookeeper/__init__.py
   M txzookeeper/client.py
   M txzookeeper/lock.py
   A txzookeeper/managed.py
   M txzookeeper/node.py
   M txzookeeper/queue.py
   A txzookeeper/retry.py
   M txzookeeper/tests/__init__.py
   M txzookeeper/tests/common.py
   A txzookeeper/tests/proxy.py
   M txzookeeper/tests/test_client.py
   A txzookeeper/tests/test_conn_failure.py
   M txzookeeper/tests/test_lock.py
   A txzookeeper/tests/test_managed.py
   M txzookeeper/tests/test_node.py
   M txzookeeper/tests/test_queue.py
   A txzookeeper/tests/test_retry.py
   M txzookeeper/tests/test_security.py
   M txzookeeper/tests/test_session.py
   M txzookeeper/tests/test_utils.py
   M txzookeeper/tests/utils.py
   M txzookeeper/todo.txt
   M txzookeeper/utils.py

[Kapil Thangavelu (hazmat)]

Please take a look.

https://codereview.appspot.com/7307051/

[lp:~hazmat/txzookeeper/backoff-retry-managed-sanity]

52. By Kapil Thangavelu on 2013-02-07

much better session recovery and retry handling

[Kapil Thangavelu (hazmat)]

Please take a look.

https://codereview.appspot.com/7307051/

[Kapil Thangavelu (hazmat)]

for reasons unknown the diff on this very wrong, all the other revs other than 52 are already on trunk.

[Benjamin Saller (bcsaller)]

Hi,

Thanks for the branch. Initially I had an older version of pyzookeeper in my path and the tests were segfaulting, but after tracking that down I was able to get things working smoothly.

Coverage shows that the session backoff code isn't tested, if I'm reading it properly self._backoff_seconds is never incremented in the tests and so some of the focus of this branch goes without regression testing.

It looks like retry_error_callback never triggers cb_retry_expired in the tests either.

Nice simplification in node.py.

This looks good but I would like to see a few more tests around the retry path.

[lp:~hazmat/txzookeeper/backoff-retry-managed-sanity]

53. By Kapil Thangavelu on 2013-02-07

app errors don't trigger reconnect

[Kapil Thangavelu (hazmat)]

the test for backoff are in test_session which operates a full zk cluster
and has a test case that verifies backoff has run. there's a new makefile
with one target .. make coverage that runs through the full test suite. in
testing with juju environments, i discovered another minor issue with
retry.

thanks for the review.

-k

On Wed, Feb 6, 2013 at 7:15 PM, Benjamin Saller <
<email address hidden>> wrote:

> Hi,
>
> Thanks for the branch. Initially I had an older version of pyzookeeper in
> my path and the tests were segfaulting, but after tracking that down I was
> able to get things working smoothly.
>
> Coverage shows that the session backoff code isn't tested, if I'm reading
> it properly self._backoff_seconds is never incremented in the tests and so
> some of the focus of this branch goes without regression testing.
>
> It looks like retry_error_callback never triggers cb_retry_expired in the
> tests either.
>
> Nice simplification in node.py.
>
> This looks good but I would like to see a few more tests around the retry
> path.
>
>
>
> --
>
> https://code.launchpad.net/~hazmat/txzookeeper/backoff-retry-managed-sanity/+merge/146938
> You are the owner of lp:~hazmat/txzookeeper/backoff-retry-managed-sanity.
>

[lp:~hazmat/txzookeeper/backoff-retry-managed-sanity]

54. By Kapil Thangavelu on 2013-02-07

argh.. conditional fail

55. By Kapil Thangavelu on 2013-02-08

prevent forced reconnect hurds

[Mikael Uvebrandt (mikael-uvebrandt)]

Would love to see this in a release soon - we've seen this reconnect hammering in production, and it's conceivable that it contributed to killing the zookeeper (3.3.3) cluster in one instance (open files limit hit).

[lp:~hazmat/txzookeeper/backoff-retry-managed-sanity]

56. By Kapil Thangavelu on 2013-05-06

incr rev

57. By Kapil Thangavelu on 2013-05-15

fix typo in reconnect test assert, cleanup hurd logic

58. By Kapil Thangavelu on 2013-05-15

work around libzk bug to make session expiration tests more reliable

[Kapil Thangavelu (hazmat)]

Merged to trunk.

=== added file '.bzrignore'
--- .bzrignore	1970-01-01 00:00:00 +0000
+++ .bzrignore	2013-05-15 13:35:05 +0000
@@ -0,0 +1,4 @@
+.emacs.desktop
+_trial_temp
+txzookeeper
+txzookeeper.egg-info

=== added file 'Makefile'
--- Makefile	1970-01-01 00:00:00 +0000
+++ Makefile	2013-05-15 13:35:05 +0000
@@ -0,0 +1,6 @@
+COVERAGE_FILES=`find txzookeeper -name "*py" | grep -v "tests"`
+
+coverage:
+	python -c "import coverage as c; c.main()" run /usr/bin/trial txzookeeper
+	python -c "import coverage as c; c.main()" html -d htmlcov $(COVERAGE_FILES)
+	gnome-open htmlcov/index.html
\ No newline at end of file

=== modified file 'debian/changelog'
--- debian/changelog	2011-07-11 21:29:44 +0000
+++ debian/changelog	2013-05-15 13:35:05 +0000
@@ -1,3 +1,21 @@
+txzookeeper (0.9.8-0ubuntu1) raring; urgency=low
+
+  * Better reconnect/retry behavior.
+
+ -- kapil <[e-mail]>  Mon, 6 May 2013 08:54:08 -0400
+
+txzookeeper (0.9.7-0ubuntu1) quantal; urgency=low
+
+  * Path included on exceptions, connection close bugfix
+
+ -- kapil <[e-mail]>  Wed, 19 Sep 2012 08:54:08 -0400
+
+txzookeeper (0.9.6-0ubuntu3) quantal; urgency=low
+
+  * managed client implementation with auto session expiration behavior
+
+ -- kapil <[e-mail]>  Fri, 22 Jun 2012 03:00:08 -0400
+
 txzookeeper (0.8.0-0ubuntu1) natty; urgency=low
 
   * Much improved session event and connection error handling (and cluster tests)

=== modified file 'setup.py'
--- setup.py	2011-06-17 23:52:34 +0000
+++ setup.py	2013-05-15 13:35:05 +0000
@@ -3,6 +3,9 @@
 #
 #  This file is part of txzookeeper.
 #
+#  Authors:
+#   Kapil Thangavelu
+#
 #  txzookeeper is free software: you can redistribute it and/or modify
 #  it under the terms of the GNU Lesser General Public License as published by
 #  the Free Software Foundation, either version 3 of the License, or
@@ -42,8 +45,8 @@
     name="txzookeeper",
     version=version,
     description="Twisted api for Apache Zookeeper",
-    author="Ensemble Developers",
-    author_email="[e-mail]",
+    author="Juju Developers",
+    author_email="[e-mail]",
     url="https://launchpad.net/txzookeeper",
     license="LGPL",
     packages=find_packages(),
@@ -55,6 +58,6 @@
         "Intended Audience :: Information Technology",
         "Programming Language :: Python",
         "Topic :: Database",
-        "License :: OSI Approved :: MIT License",
+        "License :: OSI Approved :: GNU Library or Lesser General Public License (LGPL)"
        ],
     )

=== modified file 'txzookeeper/__init__.py'
--- txzookeeper/__init__.py	2011-07-06 02:10:17 +0000
+++ txzookeeper/__init__.py	2013-05-15 13:35:05 +0000
@@ -1,8 +1,11 @@
 #
-#  Copyright (C) 2010, 2011 Canonical Ltd. All Rights Reserved
+#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
 #
 #  This file is part of txzookeeper.
 #
+#  Authors:
+#   Kapil Thangavelu
+#
 #  txzookeeper is free software: you can redistribute it and/or modify
 #  it under the terms of the GNU Lesser General Public License as published by
 #  the Free Software Foundation, either version 3 of the License, or
@@ -22,4 +25,4 @@
 __all__ = ["ZookeeperClient"]
 
 # Remember to update debian/changelog as well, for the daily build.
-version = "0.8.0"
+version = "0.9.8"

=== modified file 'txzookeeper/client.py'
--- txzookeeper/client.py	2011-06-30 20:16:17 +0000
+++ txzookeeper/client.py	2013-05-15 13:35:05 +0000
@@ -1,8 +1,11 @@
 #
-#  Copyright (C) 2010, 2011 Canonical Ltd. All Rights Reserved
+#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
 #
 #  This file is part of txzookeeper.
 #
+#  Authors:
+#   Kapil Thangavelu
+#
 #  txzookeeper is free software: you can redistribute it and/or modify
 #  it under the terms of the GNU Lesser General Public License as published by
 #  the Free Software Foundation, either version 3 of the License, or
@@ -33,6 +36,9 @@
     "scheme": "world",
     "id": "anyone"}
 
+# Skip acls for policy objects @ a higher level
+SKIP_ACLS = object()
+
 # Map result codes to exceptions classes.
 ERROR_MAPPING = {
     zookeeper.APIERROR: zookeeper.ApiErrorException,
@@ -67,6 +73,7 @@
     zookeeper.CONNECTED_STATE: "connected",
     zookeeper.CONNECTING_STATE: "connecting",
     zookeeper.EXPIRED_SESSION_STATE: "expired",
+    None: "unknown",
 }
 
 # Mapping of event type to human string.
@@ -76,7 +83,9 @@
     zookeeper.CREATED_EVENT: "created",
     zookeeper.DELETED_EVENT: "deleted",
     zookeeper.CHANGED_EVENT: "changed",
-    zookeeper.CHILD_EVENT: "child"}
+    zookeeper.CHILD_EVENT: "child",
+    None: "unknown",
+    }
 
 
 class NotConnectedException(zookeeper.ZooKeeperException):
@@ -99,9 +108,14 @@
     def type_name(self):
         return TYPE_NAME_MAPPING[self.args[1]]
 
+    @property
+    def handle(self):
+        return self.args[3]
+
     def __str__(self):
-        return "<txzookeeper.ConnectionException type: %s state: %s>" % (
-            self.type_name, self.state_name)
+        return (
+            "<txzookeeper.ConnectionException handle: %s type: %s state: %s>"
+            % (self.handle, self.type_name, self.state_name))
 
 
 def is_connection_exception(e):
@@ -122,7 +136,8 @@
     """
 
 
-class ClientEvent(namedtuple("ClientEvent", 'type, connection_state, path')):
+class ClientEvent(namedtuple("ClientEvent",
+                             'type, connection_state, path, handle')):
     """
     A client event is returned when a watch deferred fires. It denotes
     some event on the zookeeper client that the watch was requested on.
@@ -137,8 +152,8 @@
         return STATE_NAME_MAPPING[self.connection_state]
 
     def __repr__(self):
-        return  "<ClientEvent %s at %r state: %s>" % (
-            self.type_name, self.path, self.state_name)
+        return  "<ClientEvent %s at %r state: %s handle:%s>" % (
+            self.type_name, self.path, self.state_name, self.handle)
 
 
 class ZookeeperClient(object):
@@ -166,12 +181,21 @@
         self.connected = False
         self.handle = None
 
+    def __repr__(self):
+        if not self.client_id:
+            session_id = ""
+        else:
+            session_id = self.client_id[0]
+
+        return  "<%s session: %s handle: %r state: %s>" % (
+            self.__class__.__name__, session_id, self.handle, self.state)
+
     def _check_connected(self, d):
         if not self.connected:
             d.errback(NotConnectedException("not connected"))
             return d
 
-    def _check_result(self, result_code, deferred, extra_codes=()):
+    def _check_result(self, result_code, deferred, extra_codes=(), path=None):
         """Check an API call or result for errors.
 
         :param result_code: The api result code.
@@ -184,18 +208,18 @@
         error = None
         if not result_code == zookeeper.OK and not result_code in extra_codes:
             error_msg = zookeeper.zerror(result_code)
+            if path is not None:
+                error_msg += " %s" % path
             error_class = ERROR_MAPPING.get(
                 result_code, zookeeper.ZooKeeperException)
             error = error_class(error_msg)
 
             if is_connection_exception(error):
-                # Mark the client as disconnected.
-                self.connected = False
                 # Route connection errors to a connection level error
                 # handler if specified.
                 if self._connection_error_callback:
                     # The result of the connection error handler is returned
-                    # to the api.
+                    # to the api invoker.
                     d = defer.maybeDeferred(
                         self._connection_error_callback,
                         self, error)
@@ -212,14 +236,14 @@
             return d
 
         def _cb_get(result_code, value, stat):
-            if self._check_result(result_code, d):
+            if self._check_result(result_code, d, path=path):
                 return
             d.callback((value, stat))
 
         callback = self._zk_thread_callback(_cb_get)
-        watcher = self._wrap_watcher(watcher)
+        watcher = self._wrap_watcher(watcher, "get", path)
         result = zookeeper.aget(self.handle, path, watcher, callback)
-        self._check_result(result, d)
+        self._check_result(result, d, path=path)
         return d
 
     def _get_children(self, path, watcher):
@@ -228,14 +252,14 @@
             return d
 
         def _cb_get_children(result_code, children):
-            if self._check_result(result_code, d):
+            if self._check_result(result_code, d, path=path):
                 return
             d.callback(children)
 
         callback = self._zk_thread_callback(_cb_get_children)
-        watcher = self._wrap_watcher(watcher)
+        watcher = self._wrap_watcher(watcher, "child", path)
         result = zookeeper.aget_children(self.handle, path, watcher, callback)
-        self._check_result(result, d)
+        self._check_result(result, d, path=path)
         return d
 
     def _exists(self, path, watcher):
@@ -245,17 +269,17 @@
 
         def _cb_exists(result_code, stat):
             if self._check_result(
-                result_code, d, extra_codes=(zookeeper.NONODE,)):
+                result_code, d, extra_codes=(zookeeper.NONODE,), path=path):
                 return
             d.callback(stat)
 
         callback = self._zk_thread_callback(_cb_exists)
-        watcher = self._wrap_watcher(watcher)
+        watcher = self._wrap_watcher(watcher, "exists", path)
         result = zookeeper.aexists(self.handle, path, watcher, callback)
-        self._check_result(result, d)
+        self._check_result(result, d, path=path)
         return d
 
-    def _wrap_watcher(self, watcher):
+    def _wrap_watcher(self, watcher, watch_type, path):
         if watcher is None:
             return watcher
         if not callable(watcher):
@@ -273,18 +297,31 @@
         if event_type == zookeeper.SESSION_EVENT:
             if self._session_event_callback:
                 self._session_event_callback(
-                    self, ClientEvent(event_type, conn_state, path))
+                    self, ClientEvent(
+                        event_type, conn_state, path, self.handle))
+            # We do propagate to watch deferreds, in one case in
+            # particular, namely if the session is expired, in which
+            # case the watches are dead, and we send an appropriate
+            # error.
+            if conn_state == zookeeper.EXPIRED_SESSION_STATE:
+                error = zookeeper.SessionExpiredException("Session expired")
+                return watcher(None, None, None, error=error)
         else:
             return watcher(event_type, conn_state, path)
 
-    def _zk_thread_callback(self, func):
+    def _zk_thread_callback(self, func, *f_args, **f_kw):
         """
         The client library invokes callbacks in a separate thread, we wrap
         any user defined callback so that they are called back in the main
         thread after, zookeeper calls the wrapper.
         """
+        f_args = list(f_args)
+
         def wrapper(handle, *args):  # pragma: no cover
-            reactor.callFromThread(func, *args)
+            # make a copy, the conn watch callback gets invoked multiple times
+            cb_args = list(f_args)
+            cb_args.extend(args)
+            reactor.callFromThread(func, *cb_args, **f_kw)
         return wrapper
 
     @property
@@ -298,7 +335,8 @@
     @property
     def session_timeout(self):
         """
-        What's the negotiated session timeout for this connection, in seconds.
+        The negotiated session timeout for this connection, in milliseconds.
+
         If the client is not connected the value is None.
         """
         if self.connected:
@@ -326,7 +364,11 @@
         """
         if self.handle is None:
             return None
-        return zookeeper.client_id(self.handle)
+        try:
+            return zookeeper.client_id(self.handle)
+        # Invalid handle
+        except zookeeper.ZooKeeperException:
+            return None
 
     @property
     def unrecoverable(self):
@@ -334,7 +376,11 @@
         Boolean value representing whether the current connection can be
         recovered.
         """
-        return bool(zookeeper.is_unrecoverable(self.handle))
+        try:
+            return bool(zookeeper.is_unrecoverable(self.handle))
+        except zookeeper.ZooKeeperException:
+            # guard against invalid handles
+            return True
 
     def add_auth(self, scheme, identity):
         """Adds an authentication identity to this connection.
@@ -368,15 +414,21 @@
         @param force: boolean, require the connection to be closed now or
                       an exception be raised.
         """
-        if not self.connected:
-            return
-
-        result = zookeeper.close(self.handle)
         self.connected = False
+
+        if self.handle is None:
+            return
+
+        try:
+            result = zookeeper.close(self.handle)
+        except zookeeper.ZooKeeperException:
+            self.handle = None
+            return
+
         d = defer.Deferred()
-
         if self._check_result(result, d):
             return d
+        self.handle = None
         d.callback(True)
         return d
 
@@ -401,6 +453,12 @@
 
         # Use a scheduled function to ensure a timeout.
         def _check_timeout():
+            # Close the handle
+            try:
+                if self.handle is not None:
+                    zookeeper.close(self.handle)
+            except zookeeper.ZooKeeperException:
+                pass
             d.errback(
                 ConnectionTimeoutException("could not connect before timeout"))
 
@@ -416,14 +474,13 @@
         if servers is not None:
             self._servers = servers
 
-        # Assemble client id if specified.
+        # Use client id if specified.
         if client_id:
             self.handle = zookeeper.init(
                 self._servers, callback, self._session_timeout, client_id)
         else:
             self.handle = zookeeper.init(
                 self._servers, callback, self._session_timeout)
-
         return d
 
     def _cb_connected(
@@ -446,15 +503,20 @@
             # If we timed out and then connected, then close the conn.
             if state == zookeeper.CONNECTED_STATE and scheduled_timeout.called:
                 self.close()
+                self.handle = -1
+                return
 
             # Send session events to the callback, in addition to any
             # duplicate session events that will be sent for extant watches.
             if self._session_event_callback:
                 self._session_event_callback(
-                    self, ClientEvent(type, state, path))
+                    self, ClientEvent(type, state, path, self.handle))
 
             return
-        elif state == zookeeper.CONNECTED_STATE:
+        # Connected successfully, or If we're expired on an initial
+        # connect, someone else expired us.
+        elif state in (zookeeper.CONNECTED_STATE,
+                       zookeeper.EXPIRED_SESSION_STATE):
             connect_deferred.callback(self)
             return
 
@@ -470,21 +532,25 @@
         @params acls: A list of dictionaries specifying permissions.
         @params flags: Node creation flags (ephemeral, sequence, persistent)
         """
+        if acls == SKIP_ACLS:
+            acls = [ZOO_OPEN_ACL_UNSAFE]
+
         d = defer.Deferred()
         if self._check_connected(d):
             return d
 
-        def _cb_created(result_code, path):
-            if self._check_result(result_code, d):
-                return
-            d.callback(path)
-
-        callback = self._zk_thread_callback(_cb_created)
+        callback = self._zk_thread_callback(
+            self._cb_created, d, data, acls, flags)
         result = zookeeper.acreate(
             self.handle, path, data, acls, flags, callback)
-        self._check_result(result, d)
+        self._check_result(result, d, path=path)
         return d
 
+    def _cb_created(self, d, data, acls, flags, result_code, path):
+        if self._check_result(result_code, d, path=path):
+            return
+        d.callback(path)
+
     def delete(self, path, version=-1):
         """
         Delete the node at the given path. If the current node version on the
@@ -496,19 +562,17 @@
         @param version: the integer version of the node.
         """
         d = defer.Deferred()
-        if self._check_connected(d):
-            return d
-
-        def _cb_delete(result_code):
-            if self._check_result(result_code, d):
-                return
-            d.callback(result_code)
-
-        callback = self._zk_thread_callback(_cb_delete)
+
+        callback = self._zk_thread_callback(self._cb_deleted, d, path)
         result = zookeeper.adelete(self.handle, path, version, callback)
-        self._check_result(result, d)
+        self._check_result(result, d, path=path)
         return d
 
+    def _cb_deleted(self, d, path, result_code):
+        if self._check_result(result_code, d, path=path):
+            return
+        d.callback(result_code)
+
     def exists(self, path):
         """
         Check that the given node path exists. Returns a deferred that
@@ -531,8 +595,12 @@
         """
         d = defer.Deferred()
 
-        def watcher(*args):
-            d.callback(ClientEvent(*args))
+        def watcher(event_type, conn_state, path, error=None):
+            if error:
+                d.errback(error)
+            else:
+                d.callback(ClientEvent(
+                    event_type, conn_state, path, self.handle))
         return self._exists(path, watcher), d
 
     def get(self, path):
@@ -556,8 +624,12 @@
         """
         d = defer.Deferred()
 
-        def watcher(*args):
-            d.callback(ClientEvent(*args))
+        def watcher(event_type, conn_state, path, error=None):
+            if error:
+                d.errback(error)
+            else:
+                d.callback(ClientEvent(
+                    event_type, conn_state, path, self.handle))
         return self._get(path, watcher), d
 
     def get_children(self, path):
@@ -580,8 +652,12 @@
         """
         d = defer.Deferred()
 
-        def watcher(*args):
-            d.callback(ClientEvent(*args))
+        def watcher(event_type, conn_state, path, error=None):
+            if error:
+                d.errback(error)
+            else:
+                d.callback(ClientEvent(
+                    event_type, conn_state, path, self.handle))
         return self._get_children(path, watcher), d
 
     def get_acl(self, path):
@@ -598,13 +674,13 @@
             return d
 
         def _cb_get_acl(result_code, acls, stat):
-            if self._check_result(result_code, d):
+            if self._check_result(result_code, d, path=path):
                 return
             d.callback((acls, stat))
 
         callback = self._zk_thread_callback(_cb_get_acl)
         result = zookeeper.aget_acl(self.handle, path, callback)
-        self._check_result(result, d)
+        self._check_result(result, d, path=path)
         return d
 
     def set_acl(self, path, acls, version=-1):
@@ -634,17 +710,17 @@
         if self._check_connected(d):
             return d
 
-        def _cb_set_acl(result_code):
-            if self._check_result(result_code, d):
-                return
-            d.callback(result_code)
-
-        callback = self._zk_thread_callback(_cb_set_acl)
+        callback = self._zk_thread_callback(self._cb_set_acl, d, path, acls)
         result = zookeeper.aset_acl(
             self.handle, path, version, acls, callback)
-        self._check_result(result, d)
+        self._check_result(result, d, path=path)
         return d
 
+    def _cb_set_acl(self, d, path, acls, result_code):
+        if self._check_result(result_code, d, path=path):
+            return
+        d.callback(result_code)
+
     def set(self, path, data="", version=-1):
         """
         Sets the data of a node at the given path. If the current node version
@@ -660,16 +736,16 @@
         if self._check_connected(d):
             return d
 
-        def _cb_set(result_code, node_stat):
-            if self._check_result(result_code, d):
-                return
-            d.callback(node_stat)
-
-        callback = self._zk_thread_callback(_cb_set)
+        callback = self._zk_thread_callback(self._cb_set, d, path, data)
         result = zookeeper.aset(self.handle, path, data, version, callback)
-        self._check_result(result, d)
+        self._check_result(result, d, path=path)
         return d
 
+    def _cb_set(self, d, path, data, result_code, node_stat):
+        if self._check_result(result_code, d, path=path):
+            return
+        d.callback(node_stat)
+
     def set_connection_watcher(self, watcher):
         """
         Sets a permanent global watcher on the connection. This will get
@@ -679,7 +755,7 @@
         """
         if not callable(watcher):
             raise SyntaxError("Invalid Watcher %r" % (watcher))
-        watcher = self._wrap_watcher(watcher)
+        watcher = self._wrap_watcher(watcher, None, None)
         zookeeper.set_watcher(self.handle, watcher)
 
     def set_session_callback(self, callback):
@@ -717,9 +793,13 @@
         """
         if not callable(callback):
             raise TypeError("Invalid callback %r" % callback)
+        if self._connection_error_callback is not None:
+            raise RuntimeError((
+                "Connection error handlers can't be changed %s" %
+                self._connection_error_callback))
         self._connection_error_callback = callback
 
-    def set_determinstic_order(self, boolean):
+    def set_deterministic_order(self, boolean):
         """
         The zookeeper client will by default randomize the server hosts
         it will connect to unless this is set to True.
@@ -738,11 +818,11 @@
             return d
 
         def _cb_sync(result_code, path):
-            if self._check_result(result_code, d):
+            if self._check_result(result_code, d, path=path):
                 return
             d.callback(path)
 
         callback = self._zk_thread_callback(_cb_sync)
         result = zookeeper.async(self.handle, path, callback)
-        self._check_result(result, d)
+        self._check_result(result, d, path=path)
         return d

=== modified file 'txzookeeper/lock.py'
--- txzookeeper/lock.py	2011-06-17 23:52:34 +0000
+++ txzookeeper/lock.py	2013-05-15 13:35:05 +0000
@@ -1,8 +1,11 @@
 #
-#  Copyright (C) 2010 Canonical Ltd. All Rights Reserved
+#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
 #
 #  This file is part of txzookeeper.
 #
+#  Authors:
+#   Kapil Thangavelu
+#
 #  txzookeeper is free software: you can redistribute it and/or modify
 #  it under the terms of the GNU Lesser General Public License as published by
 #  the Free Software Foundation, either version 3 of the License, or
@@ -18,7 +21,7 @@
 #
 
 import zookeeper
-from twisted.internet.defer import Deferred, fail
+from twisted.internet.defer import fail
 
 
 class LockError(Exception):
@@ -133,7 +136,7 @@
         """Release the lock."""
 
         if not self._acquired:
-            error = LockError("Not holding lock %s"%(self.path))
+            error = LockError("Not holding lock %s" % (self.path))
             return fail(error)
 
         d = self._client.delete(self._candidate_path)

=== added file 'txzookeeper/managed.py'
--- txzookeeper/managed.py	1970-01-01 00:00:00 +0000
+++ txzookeeper/managed.py	2013-05-15 13:35:05 +0000
@@ -0,0 +1,413 @@
+
+from functools import partial
+
+import contextlib
+import logging
+import time
+import zookeeper
+
+from twisted.internet.defer import (
+    inlineCallbacks, DeferredLock, fail, returnValue, Deferred)
+
+from client import (
+    ZookeeperClient, ClientEvent, NotConnectedException,
+    ConnectionTimeoutException)
+from retry import RetryClient, is_session_error
+from utils import sleep
+
+
+class StopWatcher(Exception):
+    pass
+
+BACKOFF_INCREMENT = 10
+MAX_BACKOFF = 360
+
+WATCH_KIND_MAP = {
+    "child": "get_children_and_watch",
+    "exists": "exists_and_watch",
+    "get": "get_and_watch"}
+
+
+log = logging.getLogger("txzk.managed")
+#log.setLevel(logging.INFO)
+
+
+class Watch(object):
+    """
+    For application driven persistent watches, where the application
+    is manually resetting the watch.
+    """
+
+    __slots__ = ("_mgr", "_client", "_path", "_kind", "_callback")
+
+    def __init__(self, mgr, path, kind, callback):
+        self._mgr = mgr
+        self._path = path
+        self._kind = kind
+        self._callback = callback
+
+    @property
+    def path(self):
+        return self._path
+
+    @property
+    def kind(self):
+        return self._kind
+
+    @contextlib.contextmanager
+    def _ctx(self):
+        mgr = self._mgr
+        del self._mgr
+        try:
+            yield mgr
+        finally:
+            mgr.remove(self)
+
+    @inlineCallbacks
+    def reset(self):
+        with self._ctx():
+            yield self._callback(
+                zookeeper.SESSION_EVENT,
+                zookeeper.CONNECTED_STATE,
+                self._path)
+
+    def __call__(self, *args, **kw):
+        with self._ctx():
+            return self._callback(*args, **kw)
+
+    def __str__(self):
+        return "<Watcher %s %s %r>" % (self.kind, self.path, self._callback)
+
+
+class WatchManager(object):
+
+    watch_class = Watch
+
+    def __init__(self):
+        self._watches = []
+
+    def add(self, path, watch_type, watcher):
+        w = self.watch_class(self, path, watch_type, watcher)
+        self._watches.append(w)
+        return w
+
+    def remove(self, w):
+        try:
+            self._watches.remove(w)
+        except ValueError:
+            pass
+
+    def iterkeys(self):
+        for w in self._watches:
+            yield (w.path, w.kind)
+
+    def clear(self):
+        del self._watches
+        self._watches = []
+
+    @inlineCallbacks
+    def reset(self, *ignored):
+        watches = self._watches
+        self._watches = []
+
+        for w in watches:
+            try:
+                yield w.reset()
+            except Exception, e:
+                log.error("Error reseting watch %s with session event. %s %r",
+                          w, e, e)
+                continue
+
+
+class SessionClient(ZookeeperClient):
+    """A managed client that automatically re-establishes ephemerals and
+    triggers watches after reconnecting post session expiration.
+
+    This abstracts the client from session expiration handling. It does
+    come at a cost though.
+
+    There are two application constraints that need to be considered for usage
+    of the SessionClient or ManagedClient. The first is that watch callbacks
+    which examine the event, must be able to handle the synthetic session
+    event which is sent to them when the session is re-established.
+
+    The second and more problematic is that algorithms/patterns
+    utilizing ephemeral sequence nodes need to be rethought, as the
+    session client will recreate the nodes when reconnecting at their
+    previous paths. Some algorithms (like the std zk lock recipe) rely
+    on properties like the smallest valued ephemeral sequence node in
+    a container to identify the lock holder, with the notion that upon
+    session expiration a new lock/leader will be sought. Sequence
+    ephemeral node recreation in this context is problematic as the
+    node is recreated at the exact previous path. Alternative lock
+    strategies that do work are fairly simple at low volume, such as
+    owning a particular node path (ie. /locks/lock-holder) with an
+    ephemeral.
+
+    As a result the session client only tracks and restablishes non sequence
+    ephemeral nodes. For coordination around ephemeral sequence nodes it
+    provides for watching for the establishment of new sessions via
+    `subscribe_new_session`
+    """
+
+    def __init__(self, servers=None, session_timeout=None,
+                 connect_timeout=4000):
+        """
+        """
+        super(SessionClient, self).__init__(servers, session_timeout)
+        self._connect_timeout = connect_timeout
+        self._watches = WatchManager()
+        self._ephemerals = {}
+        self._session_notifications = []
+        self._reconnect_lock = DeferredLock()
+        self.set_connection_error_callback(self._cb_connection_error)
+        self.set_session_callback(self._cb_session_event)
+        self._backoff_seconds = 0
+        self._last_reconnect = time.time()
+
+    def subscribe_new_session(self):
+        d = Deferred()
+        self._session_notifications.append(d)
+        return d
+
+    @inlineCallbacks
+    def cb_restablish_session(self, e=None, forced=False):
+        """Called on intercept of session expiration to create new session.
+
+        This will reconnect to zk, re-establish ephemerals, and
+        trigger watches.
+        """
+        yield self._reconnect_lock.acquire()
+        log.debug(
+            "Connection reconnect, lock acquired handle:%d", self.handle)
+
+        try:
+            # If its been explicitly closed, don't re-establish.
+            if self.handle is None:
+                log.debug("No handle, client closed")
+                return
+
+            # Don't allow forced reconnect hurds within a session.
+            if forced and (
+                    (time.time() - self._last_reconnect)
+                    < self.session_timeout / 1000.0):
+                forced = False
+
+            if not forced and not self.unrecoverable:
+                log.debug("Client already connected, allowing retry")
+                return
+            elif self.connected or self.handle >= 0:
+                self.close()
+                self.handle = -1
+
+            # Re-establish
+            yield self._cb_restablish_session().addErrback(
+                self._cb_restablish_errback, e)
+
+        except Exception, e:
+            log.error("error while re-establish %r %s" % (e, e))
+        finally:
+            log.debug("Reconnect lock released %s", self)
+            yield self._reconnect_lock.release()
+
+    @inlineCallbacks
+    def _cb_restablish_session(self):
+        """Re-establish a new session, and recreate ephemerals and watches.
+        """
+        # Reconnect
+        while 1:
+            log.debug("Reconnect loop - connect timeout %d",
+                      self._connect_timeout)
+
+            # If we have some failures, back off
+            if self._backoff_seconds:
+                log.debug("Backing off reconnect %d" % self._backoff_seconds)
+                yield sleep(self._backoff_seconds)
+
+            # The client was explicitly closed, abort reconnect.
+            if self.handle is None:
+                returnValue(self.handle)
+            try:
+                yield self.connect(timeout=self._connect_timeout)
+                log.info("Restablished connection")
+                self._last_reconnect = time.time()
+            except ConnectionTimeoutException:
+                log.debug("Timeout establishing connection, retrying...")
+            except zookeeper.ZooKeeperException, e:
+                log.exception("Error while connecting %r %s" % (e, e))
+            except Exception, e:
+                log.info("Reconnect unknown error, aborting: %s", e)
+                raise
+            else:
+                break
+
+            if self._backoff_seconds < MAX_BACKOFF:
+                self._backoff_seconds = min(
+                    self._backoff_seconds + BACKOFF_INCREMENT,
+                    MAX_BACKOFF - 1)
+
+        # Recreate ephemerals
+        items = self._ephemerals.items()
+        self._ephemerals = {}
+
+        for path, e in items:
+            try:
+                yield self.create(
+                    path, e['data'], acls=e['acls'], flags=e['flags'])
+            except zookeeper.NodeExistsException:
+                log.error("Attempt to create ephemeral node failed %r", path)
+
+        # Signal watches
+        yield self._watches.reset()
+
+        # Notify new session observers
+        notifications = self._session_notifications
+        self._session_notifications = []
+
+        # all good, reset backoff
+        self._backoff_seconds = 0
+
+        for n in notifications:
+            n.callback(True)
+
+    def _cb_restablish_errback(self, err, failure):
+        """If there's an error re-establishing the session log it.
+        """
+        log.error("Error while trying to re-establish connection %s\n%s" % (
+            err, failure))
+        return failure
+
+    @inlineCallbacks
+    def _cb_connection_error(self,  client, error):
+        """Convert session expiration to a transient connection error.
+
+        Dispatches from api usage error.
+        """
+        if not is_session_error(error):
+            raise error
+        log.debug("Connection error detected, delaying retry...")
+        yield sleep(1)
+        raise zookeeper.ConnectionLossException
+
+    # Dispatch from retry exceed session maximum
+    def cb_retry_expired(self, error):
+        log.debug("Persistent retry error, reconnecting...")
+        return self.cb_restablish_session(forced=True)
+
+    # Dispatch from connection events
+    def _cb_session_event(self, client, event):
+        if (event.type == zookeeper.SESSION_EVENT and
+                event.connection_state == zookeeper.EXPIRED_SESSION_STATE):
+            log.debug("Client session expired event, restablishing")
+            self.cb_restablish_session()
+
+    # Client connected tracker on client operations.
+    def _check_connected(self, d):
+        """Clients are automatically reconnected."""
+        if self.connected:
+            return
+
+        if self.handle is None:
+            d.errback(NotConnectedException("Connection closed"))
+            return d
+
+        log.info("Detected dead connection, reconnecting...")
+        c_d = self.cb_restablish_session()
+
+        def after_connected(client):
+            """Return a transient connection failure.
+
+            The retry client will automatically attempt to retry the operation.
+            """
+            log.debug("Reconnected, returning transient error")
+            return fail(zookeeper.ConnectionLossException("Retry"))
+
+        c_d.addCallback(after_connected)
+        c_d.chainDeferred(d)
+        return d
+
+    # Dispatch from node watches on session expiration
+    def _watch_session_wrapper(self, watcher, event_type, conn_state, path):
+        """Watch wrapper that diverts session events to a connection callback.
+        """
+        if (event_type == zookeeper.SESSION_EVENT and
+                conn_state == zookeeper.EXPIRED_SESSION_STATE):
+            if self.unrecoverable:
+                log.debug("Watch got session expired event, reconnecting...")
+                d = self.cb_restablish_session()
+                d.addErrback(self._cb_restablish_errback)
+                return d
+
+        if event_type == zookeeper.SESSION_EVENT:
+            if self._session_event_callback:
+                self._session_event_callback(
+                    self, ClientEvent(
+                        event_type, conn_state, path, self.handle))
+        else:
+            return watcher(event_type, conn_state, path)
+
+    # Track all watches
+    def _wrap_watcher(self, watcher, watch_type, path):
+        if watcher is None:
+            return watcher
+        if not callable(watcher):
+            raise SyntaxError("invalid watcher")
+
+        # handle conn watcher, separately.
+        if watch_type is None and path is None:
+            return self._zk_thread_callback(
+                self._watch_session_wrapper, watcher)
+
+        return self._zk_thread_callback(
+            partial(
+                self._watch_session_wrapper,
+                self._watches.add(path, watch_type, watcher)))
+
+    # Track ephemerals
+    def _cb_created(self, d, data, acls, flags, result_code, path):
+        if self._check_result(result_code, d, path=path):
+            return
+
+        if (flags & zookeeper.EPHEMERAL) and not (flags & zookeeper.SEQUENCE):
+            self._ephemerals[path] = dict(
+                data=data, acls=acls, flags=flags)
+
+        d.callback(path)
+
+    def _cb_deleted(self, d, path, result_code):
+        if self._check_result(result_code, d, path=path):
+            return
+
+        self._ephemerals.pop(path, None)
+        d.callback(result_code)
+
+    def _cb_set_acl(self, d, path, acls, result_code):
+        if self._check_result(result_code, d, path=path):
+            return
+
+        if path in self._ephemerals:
+            self._ephemerals[path]['acls'] = acls
+
+        d.callback(result_code)
+
+    def _cb_set(self, d, path, data, result_code, node_stat):
+        if self._check_result(result_code, d, path=path):
+            return
+
+        if path in self._ephemerals:
+            self._ephemerals[path]['data'] = data
+
+        d.callback(node_stat)
+
+
+class _ManagedClient(RetryClient):
+
+    def subscribe_new_session(self):
+        return self.client.subscribe_new_session()
+
+
+def ManagedClient(servers=None, session_timeout=None, connect_timeout=10000):
+    client = SessionClient(servers, session_timeout, connect_timeout)
+    managed_client = _ManagedClient(client)
+    managed_client.set_retry_error_callback(client.cb_retry_expired)
+    return managed_client

=== modified file 'txzookeeper/node.py'
--- txzookeeper/node.py	2011-06-17 23:52:34 +0000
+++ txzookeeper/node.py	2013-05-15 13:35:05 +0000
@@ -1,8 +1,11 @@
 #
-#  Copyright (C) 2010 Canonical Ltd. All Rights Reserved
+#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
 #
 #  This file is part of txzookeeper.
 #
+#  Authors:
+#   Kapil Thangavelu
+#
 #  txzookeeper is free software: you can redistribute it and/or modify
 #  it under the terms of the GNU Lesser General Public License as published by
 #  the Free Software Foundation, either version 3 of the License, or
@@ -127,16 +130,9 @@
         Returns a boolean based on the node's existence. Also returns a
         deferred that fires when the node is modified/created/added/deleted.
         """
-        node_changed = Deferred()
-
-        def on_node_event((event, state, path)):
-            return node_changed.callback(
-                NodeEvent(event, state, self))
-
         d, w = self._context.exists_and_watch(self.path)
-        w.addCallback(on_node_event)
         d.addCallback(self._on_exists_success)
-        return d, node_changed
+        return d, w
 
     def _on_get_node_error(self, failure):
         failure.trap(NoNodeException)
@@ -162,17 +158,10 @@
         Retrieve the node's data and a deferred that fires when this data
         changes.
         """
-        node_changed = Deferred()
-
-        def on_node_change((event, status, path)):
-            node_changed.callback(
-                NodeEvent(event, status, self))
-
         d, w = self._context.get_and_watch(self.path)
-        w.addCallback(on_node_change)
         d.addCallback(self._on_get_node_success)
         d.addErrback(self._on_get_node_error)
-        return d, node_changed
+        return d, w
 
     def set_data(self, data):
         """Set the node's data."""
@@ -233,17 +222,9 @@
         a deferred that fires if a child is added or deleted. Optionally
         a name prefix may be passed which the child node must abide.
         """
-        children_changed = Deferred()
-
-        def on_child_added_removed((event, status, path)):
-            # path is the container not the child.
-            children_changed.callback(
-                NodeEvent(event, status, self))
-
         d, w = self._context.get_children_and_watch(self.path)
-        w.addCallback(on_child_added_removed)
         d.addCallback(self._on_get_children_filter_results, prefix)
-        return d, children_changed
+        return d, w
 
     def __cmp__(self, other):
         return cmp(self.path, other.path)

=== modified file 'txzookeeper/queue.py'
--- txzookeeper/queue.py	2011-06-30 17:14:06 +0000
+++ txzookeeper/queue.py	2013-05-15 13:35:05 +0000
@@ -1,8 +1,11 @@
 #
-#  Copyright (C) 2010 Canonical Ltd. All Rights Reserved
+#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
 #
 #  This file is part of txzookeeper.
 #
+#  Authors:
+#   Kapil Thangavelu
+#
 #  txzookeeper is free software: you can redistribute it and/or modify
 #  it under the terms of the GNU Lesser General Public License as published by
 #  the Free Software Foundation, either version 3 of the License, or
@@ -116,7 +119,7 @@
 
         flags = zookeeper.SEQUENCE
         if not self._persistent:
-            flags = flags|zookeeper.EPHEMERAL
+            flags = flags | zookeeper.EPHEMERAL
 
         d = self._client.create(
             "/".join((self._path, self.prefix)), item, self._acl, flags)

=== added file 'txzookeeper/retry.py'
--- txzookeeper/retry.py	1970-01-01 00:00:00 +0000
+++ txzookeeper/retry.py	2013-05-15 13:35:05 +0000
@@ -0,0 +1,359 @@
+#
+#  Copyright (C) 2011 Canonical Ltd. All Rights Reserved
+#
+#  This file is part of txzookeeper.
+#
+#  Authors:
+#   Kapil Thangavelu
+#
+#  txzookeeper is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU Lesser General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  txzookeeper is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public License
+#  along with txzookeeper.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+"""
+A retry client facade that transparently handles transient connection
+errors.
+"""
+
+import time
+import logging
+import zookeeper
+
+from twisted.internet.defer import inlineCallbacks, returnValue, Deferred
+
+from txzookeeper.client import NotConnectedException
+from txzookeeper.utils import sleep
+
+__all__ = ["retry", "RetryClient"]
+
+log = logging.getLogger("txzk.retry")
+
+
+class RetryException(Exception):
+    """Explicit retry exception.
+    """
+
+# Session timeout percentage that we should wait till retrying.
+RETRY_FRACTION = 30
+
+
+def is_retryable(e):
+    """Determine if an exception signifies a recoverable connection error.
+    """
+    return isinstance(
+        e,
+        (zookeeper.ClosingException,
+         zookeeper.ConnectionLossException,
+         zookeeper.OperationTimeoutException,
+         RetryException))
+
+
+def is_session_error(e):
+    return isinstance(
+        e,
+        (zookeeper.SessionExpiredException,
+         zookeeper.ConnectionLossException,
+         zookeeper.ClosingException,
+         NotConnectedException))
+
+
+def _args(args):
+    return args and args[0] or "NA"
+
+
+def get_delay(session_timeout, max_delay=5, session_fraction=RETRY_FRACTION):
+    """Get retry delay between retrying an operation.
+
+    Returns either the specified fraction of a session timeout or the
+    max delay, whichever is smaller.
+
+    The goal is to allow the connection time to auto-heal, before
+    retrying an operation.
+
+    :param session_timeout: The timeout for the session, in milliseconds
+    :param max_delay: The max delay for a retry, in seconds.
+    :param session_fraction: The fractional amount of a timeout to wait
+    """
+    retry_delay = (session_timeout * (float(session_fraction) / 100)) / 1000
+    return min(retry_delay, max_delay)
+
+
+def check_error(e):
+    """Verify a zookeeper connection error, as opposed to an app error.
+    """
+    return is_retryable(e) or is_session_error(e)
+
+
+def check_retryable(retry_client, max_time, error):
+    """Check an error and a client to see if an operation is retryable.
+
+    :param retry_client: A txzookeeper client
+    :param max_time: The max time (epoch tick) that the op is retryable till.
+    :param error: The client operation exception.
+    """
+
+    t = time.time()
+
+    # Only if the error is known.
+    if not is_retryable(error):
+        return False
+
+    # Only if we've haven't exceeded the max allotted time.
+    if max_time <= t:
+        return False
+
+    # Only if the client hasn't been explicitly closed.
+    if not retry_client.connected:
+        return False
+
+    # Only if the client is in a recoverable state.
+    if retry_client.unrecoverable:
+        return False
+
+    return True
+
+
+@inlineCallbacks
+def retry(client, func, *args, **kw):
+    """Constructs a retry wrapper around a function that retries invocations.
+
+    If the function execution results in an exception due to a transient
+    connection error, the retry wrapper will reinvoke the operation after
+    a suitable delay (fractional value of the session timeout).
+
+    :param client: A ZookeeperClient instance.
+    :param func: A callable python object that interacts with
+           zookeeper, the callable must utilize the same zookeeper
+           connection as passed in the `client` param. The function
+           must return a single value (either a deferred or result
+           value).
+    """
+    retry_started = [time.time()]
+    retry_error = False
+    while 1:
+        try:
+            value = yield func(*args, **kw)
+        except Exception, e:
+            # For clients which aren't connected (session timeout == None)
+            # we raise the errors to the callers.
+            session_timeout = client.session_timeout or 0
+
+            # The longest we keep retrying is 1.2 * session timeout
+            max_time = (session_timeout / 1000.0) * 1.2 + retry_started[0]
+
+            if not check_retryable(client, max_time, e):
+                # Check if its a persistent client error, and if so use the cb
+                # if present to try and reconnect for client errors.
+                if (check_error(e)
+                        and time.time() > max_time
+                        and callable(client.cb_retry_error)
+                        and not retry_error):
+                    log.debug("Retry error %r on %s @ %s",
+                              e, func.__name__, _args(args))
+                    retry_error = True
+                    yield client.cb_retry_error(e)
+                    retry_started[0] = time.time()
+                    continue
+                raise
+
+            # Give the connection a chance to auto-heal.
+            yield sleep(get_delay(session_timeout))
+            log.debug("Retry on %s @ %s", func.__name__, _args(args))
+            continue
+
+        returnValue(value)
+
+
+def retry_watch(client, func, *args, **kw):
+    """Contructs a wrapper around a watch callable that retries invocations.
+
+    If the callable execution results in an exception due to a transient
+    connection error, the retry wrapper will reinvoke the operation after
+    a suitable delay (fractional value of the session timeout).
+
+    A watch function must return back a tuple of deferreds
+    (value_deferred, watch_deferred). No inline callbacks are
+    performed in here to ensure that callers continue to see a
+    tuple of results.
+
+    The client passed to this retry function must be the same as
+    the one utilized by the python callable.
+
+    :param client: A ZookeeperClient instance.
+    :param func: A python callable that interacts with zookeeper. If a
+           function is passed, a txzookeeper client must the first
+           parameter of this function. The function must return a
+           tuple of (value_deferred, watch_deferred)
+    """
+    # For clients which aren't connected (session timeout == None)
+    # we raise the usage errors to the callers
+    session_timeout = client.session_timeout or 0
+
+    # If we keep retrying past the 1.2 * session timeout without
+    # success just die, the session expiry is fatal.
+    max_time = session_timeout * 1.2 + time.time()
+    value_d, watch_d = func(*args, **kw)
+
+    def retry_delay(f):
+        """Errback, verifes an op is retryable, and delays the next retry.
+        """
+        # Check that operation is retryable.
+        if not check_retryable(client, max_time, f.value):
+            return f
+
+        # Give the connection a chance to auto-heal
+        d = sleep(get_delay(session_timeout))
+        d.addCallback(retry_inner)
+
+        return d
+
+    def retry_inner(value):
+        """Retry operation invoker.
+        """
+        # Invoke the function
+        retry_value_d, retry_watch_d = func(*args, **kw)
+
+        # If we need to retry again.
+        retry_value_d.addErrback(retry_delay)
+
+        # Chain the new watch deferred to the old, presuming its doa
+        # if the value deferred errored on a connection error.
+        retry_watch_d.chainDeferred(watch_d)
+
+        # Insert back into the callback chain.
+        return retry_value_d
+
+    # Attach the retry
+    value_d.addErrback(retry_delay)
+
+    return value_d, watch_d
+
+
+def _passproperty(name):
+    """Returns a method wrapper that delegates to a client's property.
+    """
+    def wrapper(retry_client):
+        return getattr(retry_client.client, name)
+    return property(wrapper)
+
+
+class RetryClient(object):
+    """A ZookeeperClient wrapper that transparently performs retries.
+
+    A zookeeper connection can experience transient connection failures
+    on any operation. As long as the session associated to the connection
+    is still active on the zookeeper cluster, libzookeeper can reconnect
+    automatically to the cluster and session and the client is able to
+    retry.
+
+    Whether a given operation is safe for retry depends on the application
+    in question and how's interacting with zookeeper.
+
+    In particular coordination around sequence nodes can be
+    problematic, as the client has no way of knowing if the operation
+    succeed or not without additional application specific context.
+
+    Idempotent operations against the zookeeper tree are generally
+    safe to retry.
+
+    This class provides a simple wrapper around a zookeeper client,
+    that will automatically perform retries on operations that
+    interact with the zookeeper tree, in the face of transient errors,
+    till the session timeout has been reached. All of the attributes
+    and methods of a zookeeper client are exposed.
+
+    All the methods of the client that interact with the zookeeper tree
+    are retry enabled.
+    """
+
+    def __init__(self, client):
+        self.client = client
+        self.client.cb_retry_error = None
+
+    def set_retry_error_callback(self, callback):
+        self.client.cb_retry_error = callback
+
+    def add_auth(self, *args, **kw):
+        return retry(self.client, self.client.add_auth, *args, **kw)
+
+    def create(self, *args, **kw):
+        return retry(self.client, self.client.create, *args, **kw)
+
+    def delete(self, *args, **kw):
+        return retry(self.client, self.client.delete, *args, **kw)
+
+    def exists(self, *args, **kw):
+        return retry(self.client, self.client.exists, *args, **kw)
+
+    def get(self, *args, **kw):
+        return retry(self.client, self.client.get, *args, **kw)
+
+    def get_acl(self, *args, **kw):
+        return retry(self.client, self.client.get_acl, *args, **kw)
+
+    def get_children(self, *args, **kw):
+        return retry(self.client, self.client.get_children, *args, **kw)
+
+    def set_acl(self, *args, **kw):
+        return retry(self.client, self.client.set_acl, *args, **kw)
+
+    def set(self, *args, **kw):
+        return retry(self.client, self.client.set, *args, **kw)
+
+    def sync(self, *args, **kw):
+        return retry(self.client, self.client.sync, *args, **kw)
+
+    # Watch retries
+
+    def exists_and_watch(self, *args, **kw):
+        return retry_watch(
+            self.client, self.client.exists_and_watch, *args, **kw)
+
+    def get_and_watch(self, *args, **kw):
+        return retry_watch(
+            self.client, self.client.get_and_watch, *args, **kw)
+
+    def get_children_and_watch(self, *args, **kw):
+        return retry_watch(
+            self.client, self.client.get_children_and_watch, *args, **kw)
+
+    # Passthrough methods
+
+    def set_connection_watcher(self, *args, **kw):
+        return self.client.set_connection_watcher(*args, **kw)
+
+    def set_connection_error_callback(self, *args, **kw):
+        return self.client.set_connection_error_callback(*args, **kw)
+
+    def set_session_callback(self, *args, **kw):
+        return self.client.set_session_callback(*args, **kw)
+
+    def set_determinstic_order(self, *args, **kw):
+        return self.client.set_determinstic_order(*args, **kw)
+
+    def close(self):
+        return self.client.close()
+
+    @inlineCallbacks
+    def connect(self, *args, **kw):
+        yield self.client.connect(*args, **kw)
+        returnValue(self)
+
+    # passthrough properties
+    state = _passproperty("state")
+    client_id = _passproperty("client_id")
+    session_timeout = _passproperty("session_timeout")
+    servers = _passproperty("servers")
+    handle = _passproperty("handle")
+    connected = _passproperty("connected")
+    unrecoverable = _passproperty("unrecoverable")

=== modified file 'txzookeeper/tests/__init__.py'
--- txzookeeper/tests/__init__.py	2011-06-30 18:05:14 +0000
+++ txzookeeper/tests/__init__.py	2013-05-15 13:35:05 +0000
@@ -1,8 +1,11 @@
 #
-#  Copyright (C) 2010 Canonical Ltd. All Rights Reserved
+#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
 #
 #  This file is part of txzookeeper.
 #
+#  Authors:
+#   Kapil Thangavelu
+#
 #  txzookeeper is free software: you can redistribute it and/or modify
 #  it under the terms of the GNU Lesser General Public License as published by
 #  the Free Software Foundation, either version 3 of the License, or
@@ -17,9 +20,12 @@
 #  along with txzookeeper.  If not, see <http://www.gnu.org/licenses/>.
 #
 
+import logging
+import StringIO
 import sys
 
 import zookeeper
+from twisted.internet.defer import Deferred
 from twisted.trial.unittest import TestCase
 
 from mocker import MockerTestCase
@@ -29,20 +35,64 @@
 
     def setUp(self):
         super(ZookeeperTestCase, self).setUp()
-        self.log_file_path = self.makeFile()
-        self.log_file = open(self.log_file_path, 'w')
-        #zookeeper.set_log_stream(self.log_file)
         zookeeper.set_debug_level(0)
 
     def tearDown(self):
         super(ZookeeperTestCase, self).tearDown()
-        #zookeeper.set_log_stream(sys.stderr)  # reset to default
-        #zookeeper.set_debug_level(zookeeper.LOG_LEVEL_DEBUG)
-        self.log_file.close()
 
     def get_log(self):
         return open(self.log_file_path).read()
 
+    def sleep(self, delay):
+        """Non-blocking sleep."""
+        from twisted.internet import reactor
+        deferred = Deferred()
+        reactor.callLater(delay, deferred.callback, None)
+        return deferred
+
+    _missing_attr = object()
+
+    def patch(self, object, attr, value):
+        """Replace an object's attribute, and restore original value later.
+
+        Returns the original value of the attribute if any or None.
+        """
+        original_value = getattr(object, attr, self._missing_attr)
+
+        @self.addCleanup
+        def restore_original():
+            if original_value is self._missing_attr:
+                try:
+                    delattr(object, attr)
+                except AttributeError:
+                    pass
+            else:
+                setattr(object, attr, original_value)
+        setattr(object, attr, value)
+        if original_value is self._missing_attr:
+            return None
+        return original_value
+
+    def capture_log(self, name="", level=logging.INFO,
+                    log_file=None, formatter=None):
+        """Capture log channel to StringIO"""
+        if log_file is None:
+            log_file = StringIO.StringIO()
+        log_handler = logging.StreamHandler(log_file)
+        if formatter:
+            log_handler.setFormatter(formatter)
+        logger = logging.getLogger(name)
+        logger.addHandler(log_handler)
+        old_logger_level = logger.level
+        logger.setLevel(level)
+
+        @self.addCleanup
+        def reset_logging():
+            logger.removeHandler(log_handler)
+            logger.setLevel(old_logger_level)
+
+        return log_file
+
 
 def egg_test_runner():
     """

=== modified file 'txzookeeper/tests/common.py'
--- txzookeeper/tests/common.py	2011-06-18 15:35:03 +0000
+++ txzookeeper/tests/common.py	2013-05-15 13:35:05 +0000
@@ -1,3 +1,26 @@
+#
+#  Copyright (C) 2010-2011, 2011 Canonical Ltd. All Rights Reserved
+#
+#  This file is part of txzookeeper.
+#
+#  Authors:
+#   Kapil Thangavelu
+#
+#  txzookeeper is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU Lesser General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  txzookeeper is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public License
+#  along with txzookeeper.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+
 import os
 import os.path
 import shutil
@@ -46,7 +69,7 @@
 
         # various setup steps
         if not os.path.exists(self.working_path):
-            os.mdir(self.working_path)
+            os.mkdir(self.working_path)
         if not os.path.exists(log_path):
             os.mkdir(log_path)
         if not os.path.exists(data_path):
@@ -54,10 +77,12 @@
 
         with open(config_path, "w") as config:
             config.write("""
-tickTime=2000
+tickTime=1000
 dataDir=%s
 clientPort=%s
 maxClientCnxns=0
+maxSessionTimeout=5000
+minSessionTimeout=2000
 """ % (data_path, self.server_info.client_port))
 
         # setup a replicated setup if peers are specified

=== added file 'txzookeeper/tests/proxy.py'
--- txzookeeper/tests/proxy.py	1970-01-01 00:00:00 +0000
+++ txzookeeper/tests/proxy.py	2013-05-15 13:35:05 +0000
@@ -0,0 +1,97 @@
+#
+#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
+#
+#  This file is part of txzookeeper.
+#
+#  Authors:
+#   Kapil Thangavelu
+#
+#  txzookeeper is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU Lesser General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  txzookeeper is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public License
+#  along with txzookeeper.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+from twisted.protocols import portforward
+
+
+class Blockable(object):
+
+    _blocked = None
+
+    def set_blocked(self, value):
+        value = bool(value)
+        self._blocked = value
+        if self.transport and not self._blocked:
+            self.transport.resumeProducing()
+
+
+class ProxyClient(portforward.ProxyClient, Blockable):
+
+    def dataReceived(self, data):
+        if self._blocked:
+            return
+        portforward.ProxyClient.dataReceived(self, data)
+
+    def setServer(self, server):
+        server.set_blocked(self._blocked)
+        super(ProxyClient, self).setServer(server)
+
+    def connectionMade(self):
+        self.peer.setPeer(self)
+        if not self._blocked:
+            # The server waits till the client is connected
+            self.peer.transport.resumeProducing()
+        else:
+            self.transport.pauseProducing()
+
+
+class ProxyClientFactory(portforward.ProxyClientFactory):
+
+    protocol = ProxyClient
+
+
+class ProxyServer(portforward.ProxyServer, Blockable):
+
+    clientProtocolFactory = ProxyClientFactory
+
+    def dataReceived(self, data):
+        if self._blocked:
+            return
+        portforward.ProxyServer.dataReceived(self, data)
+
+
+class ProxyFactory(portforward.ProxyFactory):
+
+    protocol = ProxyServer
+    instance = _blocked = False
+
+    def lose_connection(self):
+        """Terminate both ends of the proxy connection."""
+        if self.instance:
+            self.instance.transport.loseConnection()
+        if self.instance.peer:
+            self.instance.peer.transport.loseConnection()
+
+    def set_blocked(self, value):
+        self._blocked = bool(value)
+        if self.instance:
+            self.instance.set_blocked(self._blocked)
+            if self.instance.peer:
+                self.instance.peer.set_blocked(self._blocked)
+
+    def buildProtocol(self, addr):
+        # Track last protocol used, on reconnect any pauses are disabled.
+        self.instance = portforward.ProxyFactory.buildProtocol(self, addr)
+        # Propogate the value, the client will aggressively try to
+        # reconnect else.
+        self.instance.set_blocked(self._blocked)
+        return self.instance

=== modified file 'txzookeeper/tests/test_client.py'
--- txzookeeper/tests/test_client.py	2011-06-30 20:16:17 +0000
+++ txzookeeper/tests/test_client.py	2013-05-15 13:35:05 +0000
@@ -1,8 +1,11 @@
 #
-#  Copyright (C) 2010, 2011 Canonical Ltd. All Rights Reserved
+#  Copyright (C) 2010-2011, 2011 Canonical Ltd. All Rights Reserved
 #
 #  This file is part of txzookeeper.
 #
+#  Authors:
+#   Kapil Thangavelu
+#
 #  txzookeeper is free software: you can redistribute it and/or modify
 #  it under the terms of the GNU Lesser General Public License as published by
 #  the Free Software Foundation, either version 3 of the License, or
@@ -21,13 +24,13 @@
 import base64
 import hashlib
 
-from twisted.internet.defer import Deferred
+from twisted.internet.defer import Deferred, maybeDeferred
 from twisted.internet.base import DelayedCall
 from twisted.python.failure import Failure
 
 import zookeeper
 
-from mocker import ANY, MATCH
+from mocker import ANY, MATCH, ARGS
 from txzookeeper.tests import ZookeeperTestCase, utils
 from txzookeeper.client import (
     ZookeeperClient, ZOO_OPEN_ACL_UNSAFE, ConnectionTimeoutException,
@@ -119,18 +122,55 @@
         d.addCallback(check_connected)
         return d
 
+    def test_close(self):
+        """
+        Test that the connection is closed, also for the first
+        connection when the zookeeper handle is 0.
+        """
+
+        def _fake_init(*_):
+            return 0
+
+        mock_init = self.mocker.replace("zookeeper.init")
+        mock_init(ARGS)
+        self.mocker.call(_fake_init)
+
+        def _fake_close(handle):
+            return zookeeper.OK
+
+        mock_close = self.mocker.replace("zookeeper.close")
+        mock_close(0)
+        self.mocker.call(_fake_close)
+
+        self.mocker.replay()
+
+        # Avoid unclean reactor by letting the callLater go through,
+        # but we do not care about the timeout.
+        def _silence_timeout(failure):
+            failure.trap(ConnectionTimeoutException)
+        self.client.connect(timeout=0).addErrback(_silence_timeout)
+
+        d = maybeDeferred(self.client.close)
+
+        def _verify(result):
+            self.mocker.verify()
+        d.addCallback(_verify)
+        return d
+
     def test_client_event_repr(self):
         event = ClientEvent(zookeeper.SESSION_EVENT,
-                            zookeeper.EXPIRED_SESSION_STATE, '')
-        self.assertEqual(repr(event),
-                         "<ClientEvent session at '' state: expired>")
+                            zookeeper.EXPIRED_SESSION_STATE, '', 0)
+        self.assertEqual(
+            repr(event),
+            "<ClientEvent session at '' state: expired handle:0>")
 
     def test_client_event_attributes(self):
-        event = ClientEvent(4, 'state', 'path')
+        event = ClientEvent(4, 'state', 'path', 0)
         self.assertEqual(event.type, 4)
         self.assertEqual(event.connection_state, 'state')
         self.assertEqual(event.path, 'path')
-        self.assertEqual(event, (4, 'state', 'path'))
+        self.assertEqual(event.handle, 0)
+        self.assertEqual(event, (4, 'state', 'path', 0))
 
     def test_client_use_while_disconnected_returns_failure(self):
         return self.assertFailure(
@@ -401,7 +441,7 @@
         """
         d = self.client.connect()
 
-        def inject_error(result_code, d, extra_codes=None):
+        def inject_error(result_code, d, extra_codes=None, path=None):
             error = SyntaxError()
             d.errback(error)
             return error
@@ -409,7 +449,8 @@
         def check_exists(client):
             mock_client = self.mocker.patch(client)
             mock_client._check_result(
-                ANY, DEFERRED_MATCH, extra_codes=(zookeeper.NONODE,))
+                ANY, DEFERRED_MATCH, extra_codes=(zookeeper.NONODE,),
+                path="/zebra-moon")
             self.mocker.call(inject_error)
             self.mocker.replay()
             return client.exists("/zebra-moon")
@@ -506,7 +547,7 @@
 
         def check_exists(path):
             exists, watch = self.client.exists_and_watch(
-                "%s/wooly-mammoth"%node_path)
+                "%s/wooly-mammoth" % node_path)
             watch.chainDeferred(watcher_deferred)
             return exists
 
@@ -517,13 +558,14 @@
 
         def create_node(client):
             self.assertEqual(client.connected, True)
-            return self.client2.create("%s/wooly-mammoth"%node_path, "extinct")
+            return self.client2.create(
+                "%s/wooly-mammoth" % node_path, "extinct")
 
         def shim(path):
             return watcher_deferred
 
         def verify_watch(event):
-            self.assertEqual(event.path, "%s/wooly-mammoth"%node_path)
+            self.assertEqual(event.path, "%s/wooly-mammoth" % node_path)
             self.assertEqual(event.type, zookeeper.CREATED_EVENT)
 
         d.addCallback(create_container)
@@ -600,7 +642,8 @@
 
         def verify_succeeds(failure):
             self.assertTrue(failure)
-            self.assertEqual(failure.value.args, ("no node",))
+            self.assertEqual(
+                failure.value.args, ("no node /abcd",))
 
         d.addCallback(delete_node)
         d.addCallback(verify_fails)
@@ -643,7 +686,8 @@
 
         def verify_succeeds(failure):
             self.assertTrue(failure)
-            self.assertEqual(failure.value.args, ("no node",))
+            self.assertTrue(
+                failure.value.args, ("no node /xy1"))
 
         d.addCallback(set_node)
         d.addCallback(verify_fails)
@@ -833,7 +877,7 @@
             return self.client.set("/orchard", "bar")
 
         def node_access_failed(failure):
-            self.assertEqual(failure.value.args, ("not authenticated",))
+            self.assertEqual(failure.value.args, ("not authenticated /orchard",))
             failed.append(True)
             return
 
@@ -1046,7 +1090,7 @@
         d = self.client.connect()
 
         def verify_session_timeout(client):
-            self.assertEqual(client.session_timeout, 4000)
+            self.assertIn(client.session_timeout, (4000, 10000))
 
         d.addCallback(verify_session_timeout)
         return d
@@ -1082,60 +1126,6 @@
         d.addErrback(verify_invalid)
         return d
 
-    def xtest_session_expired_event(self):
-        """
-        A client session can be reattached to in a separate connection,
-        if the a session is expired, using the zookeeper connection will
-        raise a SessionExpiredException.
-        """
-        d = self.client.connect()
-
-        class StopTest(Exception):
-            pass
-
-        def new_connection_same_connection(client):
-            self.assertEqual(client.state, zookeeper.CONNECTED_STATE)
-            return ZookeeperClient("127.0.0.1:2181").connect(
-                client_id=client.client_id).addErrback(
-                guard_session_expired, client)
-
-        def guard_session_expired(failure, client):
-            # On occassion we get a session expired event while connecting.
-            failure.trap(ConnectionException)
-            self.assertEqual(failure.value.state_name, "expired")
-            # Stop the test from proceeding
-            raise StopTest()
-
-        def close_new_connection(client):
-            # Verify both connections are using same session
-            self.assertEqual(self.client.client_id, client.client_id)
-            self.assertEqual(client.state, zookeeper.CONNECTED_STATE)
-
-            # Closing one connection will close the session
-            client.close()
-
-            # Continued use of the other client will get a
-            # disconnect exception.
-            return self.client.exists("/")
-
-        def verify_original_closed(failure):
-            if not isinstance(failure, Failure):
-                self.fail("Test did not raise exception.")
-            failure.trap(
-                zookeeper.SessionExpiredException,
-                zookeeper.ConnectionLossException)
-
-            #print "client close"
-            self.client.close()
-            #print "creating new client for teardown"
-            return self.client.connect()
-
-        d.addCallback(new_connection_same_connection)
-        d.addCallback(close_new_connection)
-        d.addBoth(verify_original_closed)
-
-        return d
-
     def test_connect_with_server(self):
         """
         A client's servers can be specified in the connect method.
@@ -1183,12 +1173,8 @@
         connected before then, then an errback is invoked with a timeout
         exception.
         """
-        mock_init = self.mocker.replace("zookeeper.init")
-        mock_init(ANY, ANY, ANY)
-        self.mocker.result(0)
-        self.mocker.replay()
-
-        d = self.client.connect(timeout=0.1)
+        # Connect to a non standard port with nothing at the remote side.
+        d = self.client.connect("127.0.0.1:2182", timeout=0.2)
 
         def verify_timeout(failure):
             self.assertTrue(

=== added file 'txzookeeper/tests/test_conn_failure.py'
--- txzookeeper/tests/test_conn_failure.py	1970-01-01 00:00:00 +0000
+++ txzookeeper/tests/test_conn_failure.py	2013-05-15 13:35:05 +0000
@@ -0,0 +1,194 @@
+#
+#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
+#
+#  This file is part of txzookeeper.
+#
+#  Authors:
+#   Kapil Thangavelu
+#
+#  txzookeeper is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU Lesser General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  txzookeeper is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public License
+#  along with txzookeeper.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+import zookeeper
+
+from twisted.internet import reactor
+from twisted.internet.defer import inlineCallbacks
+
+from txzookeeper import ZookeeperClient
+from txzookeeper.tests import ZookeeperTestCase, utils
+from txzookeeper.tests.proxy import ProxyFactory
+
+
+class WatchDeliveryConnectionFailedTest(ZookeeperTestCase):
+    """Watches are still sent on reconnect.
+    """
+
+    def setUp(self):
+        super(WatchDeliveryConnectionFailedTest, self).setUp()
+
+        self.proxy = ProxyFactory("127.0.0.1", 2181)
+        self.proxy_port = reactor.listenTCP(0, self.proxy)
+        host = self.proxy_port.getHost()
+        self.proxied_client = ZookeeperClient(
+            "%s:%s" % (host.host, host.port))
+        self.direct_client = ZookeeperClient("127.0.0.1:2181", 3000)
+        self.session_events = []
+
+        def session_event_collector(conn, event):
+            self.session_events.append(event)
+
+        self.proxied_client.set_session_callback(session_event_collector)
+        return self.direct_client.connect()
+
+    @inlineCallbacks
+    def tearDown(self):
+        zookeeper.set_debug_level(0)
+        if self.proxied_client.connected:
+            yield self.proxied_client.close()
+        if not self.direct_client.connected:
+            yield self.direct_client.connect()
+        utils.deleteTree(handle=self.direct_client.handle)
+        yield self.direct_client.close()
+        self.proxy.lose_connection()
+        yield self.proxy_port.stopListening()
+
+    def verify_events(self, events, expected):
+        """Verify the state of the session events encountered.
+        """
+        for value, state in zip([e.state_name for e in events], expected):
+            self.assertEqual(value, state)
+
+    @inlineCallbacks
+    def test_child_watch_fires_upon_reconnect(self):
+        yield self.proxied_client.connect()
+
+        # Setup tree
+        cpath = "/test-tree"
+        yield self.direct_client.create(cpath)
+
+        # Setup watch
+        child_d, watch_d = self.proxied_client.get_children_and_watch(cpath)
+
+        self.assertEqual((yield child_d), [])
+
+        # Kill the connection and fire the watch
+        self.proxy.lose_connection()
+        yield self.direct_client.create(
+            cpath + "/abc", flags=zookeeper.SEQUENCE)
+
+        # We should still get the child event.
+        yield watch_d
+
+        # We get two pairs of (connecting, connected) for the conn and watch
+        self.assertEqual(len(self.session_events), 4)
+        self.verify_events(
+            self.session_events,
+            ("connecting", "connecting", "connected", "connected"))
+
+    @inlineCallbacks
+    def test_exists_watch_fires_upon_reconnect(self):
+        yield self.proxied_client.connect()
+        cpath = "/test"
+
+        # Setup watch
+        exists_d, watch_d = self.proxied_client.exists_and_watch(cpath)
+
+        self.assertEqual((yield exists_d), None)
+
+        # Kill the connection and fire the watch
+        self.proxy.lose_connection()
+        yield self.direct_client.create(cpath)
+
+        # We should still get the exists event.
+        yield watch_d
+
+        # We get two pairs of (connecting, connected) for the conn and watch
+        self.assertEqual(len(self.session_events), 4)
+        self.verify_events(
+            self.session_events,
+            ("connecting", "connecting", "connected", "connected"))
+
+    @inlineCallbacks
+    def test_get_watch_fires_upon_reconnect(self):
+        yield self.proxied_client.connect()
+        # Setup tree
+        cpath = "/test"
+        yield self.direct_client.create(cpath, "abc")
+
+        # Setup watch
+        get_d, watch_d = self.proxied_client.get_and_watch(cpath)
+        content, stat = yield get_d
+        self.assertEqual(content, "abc")
+
+        # Kill the connection and fire the watch
+        self.proxy.lose_connection()
+        yield self.direct_client.set(cpath, "xyz")
+
+        # We should still get the exists event.
+        yield watch_d
+
+        # We also two pairs of (connecting, connected) for the conn and watch
+        self.assertEqual(len(self.session_events), 4)
+        self.verify_events(
+            self.session_events,
+            ("connecting", "connecting", "connected", "connected"))
+
+    @inlineCallbacks
+    def test_watch_delivery_failure_resends(self):
+        """Simulate a network failure for the watch delivery
+
+        The zk server effectively sends the watch delivery to the client,
+        but the client never recieves it.
+        """
+        yield self.proxied_client.connect()
+        cpath = "/test"
+
+        # Setup watch
+        exists_d, watch_d = self.proxied_client.exists_and_watch(cpath)
+
+        self.assertEqual((yield exists_d), None)
+
+        # Pause the connection fire the watch, and blackhole the data.
+        self.proxy.set_blocked(True)
+        yield self.direct_client.create(cpath)
+        self.proxy.set_blocked(False)
+        self.proxy.lose_connection()
+
+        # We should still get the exists event.
+        yield watch_d
+
+    @inlineCallbacks
+    def xtest_binding_bug_session_exception(self):
+        """This test triggers an exception in the python-zookeeper binding.
+
+        File "txzookeeper/client.py", line 491, in create
+           self.handle, path, data, acls, flags, callback)
+        exceptions.SystemError: error return without exception set
+        """
+        yield self.proxied_client.connect()
+        data_d, watch_d = yield self.proxied_client.exists_and_watch("/")
+        self.assertTrue((yield data_d))
+        self.proxy.set_blocked(True)
+        # Wait for session expiration, on a single server options are limited
+        yield self.sleep(15)
+        # Unblock the proxy for next connect, and then drop the connection.
+        self.proxy.set_blocked(False)
+        self.proxy.lose_connection()
+        # Wait for a reconnect
+        yield self.assertFailure(watch_d, zookeeper.SessionExpiredException)
+        # Leads to bindings bug failure
+        yield self.assertFailure(
+            self.proxied_client.get("/a"),
+            zookeeper.SessionExpiredException)
+        self.assertEqual(self.session_events[-1].state_name, "expired")

=== modified file 'txzookeeper/tests/test_lock.py'
--- txzookeeper/tests/test_lock.py	2011-06-17 23:52:34 +0000
+++ txzookeeper/tests/test_lock.py	2013-05-15 13:35:05 +0000
@@ -1,8 +1,11 @@
 #
-#  Copyright (C) 2010 Canonical Ltd. All Rights Reserved
+#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
 #
 #  This file is part of txzookeeper.
 #
+#  Authors:
+#   Kapil Thangavelu
+#
 #  txzookeeper is free software: you can redistribute it and/or modify
 #  it under the terms of the GNU Lesser General Public License as published by
 #  the Free Software Foundation, either version 3 of the License, or

=== added file 'txzookeeper/tests/test_managed.py'
--- txzookeeper/tests/test_managed.py	1970-01-01 00:00:00 +0000
+++ txzookeeper/tests/test_managed.py	2013-05-15 13:35:05 +0000
@@ -0,0 +1,309 @@
+#
+#  Copyright (C) 2010-2012 Canonical Ltd. All Rights Reserved
+#
+#  This file is part of txzookeeper.
+#
+#  Authors:
+#   Kapil Thangavelu
+#
+#  txzookeeper is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU Lesser General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  txzookeeper is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public License
+#  along with txzookeeper.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+import logging
+import zookeeper
+
+from twisted.internet.defer import inlineCallbacks, Deferred, DeferredList
+
+from txzookeeper.client import ZookeeperClient, ClientEvent
+from txzookeeper import managed
+from txzookeeper.tests import ZookeeperTestCase, utils
+from txzookeeper.tests import test_client
+
+
+class WatchTest(ZookeeperTestCase):
+    """Watch manager and watch tests"""
+
+    def setUp(self):
+        self.watches = managed.WatchManager()
+
+    def tearDown(self):
+        self.watches.clear()
+        del self.watches
+
+    def test_add_remove(self):
+
+        w = self.watches.add("/foobar", "child", lambda x: 1)
+        self.assertIn(
+            "<Watcher child /foobar",
+            str(w))
+        self.assertIn(w, self.watches._watches)
+        self.watches.remove(w)
+        self.assertNotIn(w, self.watches._watches)
+        self.watches.remove(w)
+
+    @inlineCallbacks
+    def test_watch_fire_removes(self):
+        """Firing the watch removes it from the manager.
+        """
+        w = self.watches.add("/foobar", "child", lambda x: 1)
+        yield w("a")
+        self.assertNotIn(w, self.watches._watches)
+
+    @inlineCallbacks
+    def test_watch_fire_with_error_removes(self):
+        """Firing the watch removes it from the manager.
+        """
+        d = Deferred()
+
+        @inlineCallbacks
+        def cb_error(e):
+            yield d
+            raise ValueError("a")
+
+        w = self.watches.add("/foobar", "child", lambda x: 1)
+        try:
+            wd = w("a")
+            d.callback(True)
+            yield wd
+        except ValueError:
+            pass
+        self.assertNotIn(w, self.watches._watches)
+
+    @inlineCallbacks
+    def test_reset_with_error(self):
+        """A callback firing an error on reset is ignored.
+        """
+        output = self.capture_log("txzk.managed")
+        d = Deferred()
+        results = []
+
+        @inlineCallbacks
+        def callback(*args, **kw):
+            results.append((args, kw))
+            yield d
+            raise ValueError("a")
+
+        w = self.watches.add("/foobar", "child", callback)
+        reset_done = self.watches.reset()
+
+        e, _ = results.pop()
+        e = list(e)
+        e.append(0)
+
+        self.assertEqual(
+            str(ClientEvent(*e)),
+            "<ClientEvent session at '/foobar' state: connected handle:0>")
+        d.callback(True)
+        yield reset_done
+        self.assertNotIn(w, self.watches._watches)
+        self.assertIn("Error reseting watch", output.getvalue())
+
+    @inlineCallbacks
+    def test_reset(self):
+        """Reset fires a synthentic client event, and clears watches.
+        """
+        d = Deferred()
+        results = []
+
+        def callback(*args, **kw):
+            results.append((args, kw))
+            return d
+
+        w = self.watches.add("/foobar", "child", callback)
+        reset_done = self.watches.reset()
+
+        e, _ = results.pop()
+        e = list(e)
+        e.append(0)
+
+        self.assertEqual(
+            str(ClientEvent(*e)),
+            "<ClientEvent session at '/foobar' state: connected handle:0>")
+        d.callback(True)
+        yield reset_done
+        self.assertNotIn(w, self.watches._watches)
+
+
+class SessionClientTests(test_client.ClientTests):
+    """Run through basic operations with SessionClient."""
+    timeout = 5
+
+    def setUp(self):
+        super(SessionClientTests, self).setUp()
+        self.client = managed.SessionClient("127.0.0.1:2181")
+
+    def test_client_use_while_disconnected_returns_failure(self):
+        # managed session client reconnects here.
+        return True
+
+
+class SessionClientExpireTests(ZookeeperTestCase):
+    """Verify expiration behavior."""
+
+    def setUp(self):
+        super(SessionClientExpireTests, self).setUp()
+        self.client = managed.ManagedClient("127.0.0.1:2181", 3000)
+        self.client2 = None
+        self.output = self.capture_log(level=logging.DEBUG)
+        return self.client.connect()
+
+    @inlineCallbacks
+    def tearDown(self):
+        self.client.close()
+        self.client2 = ZookeeperClient("127.0.0.1:2181")
+        yield self.client2.connect()
+        utils.deleteTree(handle=self.client2.handle)
+        yield self.client2.close()
+        super(SessionClientExpireTests, self).tearDown()
+
+    @inlineCallbacks
+    def expire_session(self, wait=True, retry=0):
+        assert self.client.connected
+        #if wait:
+        #    d = self.client.subscribe_new_session()
+        client_id = self.client.client_id
+        self.client2 = ZookeeperClient(self.client.servers)
+        yield self.client2.connect(client_id=client_id)
+        yield self.client2.close()
+
+        # It takes some time to propagate (1/3 session time as ping)
+        if wait:
+            yield self.sleep(2)
+            client_new_id = self.client.client_id
+            # Crappy workaround to c libzk bug/issue see http://goo.gl/9ei5c
+            # Works most of the time.. but bound it when it doesn't. lame!
+            if client_id[0] == client_new_id[0] and retry < 10:
+                yield self.expire_session(wait, retry+1)
+
+    @inlineCallbacks
+    def test_session_expiration_conn(self):
+        d = self.client.subscribe_new_session()
+        session_id = self.client.client_id[0]
+        yield self.client.create("/fo-1", "abc")
+        yield self.expire_session(wait=True)
+        yield d
+        stat = yield self.client.exists("/")
+        self.assertTrue(stat)
+        self.assertNotEqual(session_id, self.client.client_id[0])
+
+    @inlineCallbacks
+    def test_session_expiration_notification(self):
+        session_id = self.client.client_id[0]
+        c_d, w_d = self.client.get_and_watch("/")
+        yield c_d
+        d = self.client.subscribe_new_session()
+        self.assertFalse(d.called)
+        yield self.expire_session(wait=True)
+        yield d
+        yield w_d
+        self.assertNotEqual(session_id, self.client.client_id[0])
+
+    @inlineCallbacks
+    def test_invoked_watch_gc(self):
+        c_d, w_d = yield self.client.get_children_and_watch("/")
+        yield c_d
+        yield self.client.create("/foo")
+        yield w_d
+        yield self.expire_session()
+        yield self.client.create("/foo2")
+        # Nothing should blow up
+        yield self.sleep(0.2)
+
+    @inlineCallbacks
+    def test_app_usage_error_bypass_retry(self):
+        """App errors shouldn't trigger a reconnect."""
+        output = self.capture_log(level=logging.DEBUG)
+        yield self.assertFailure(
+            self.client.get("/abc"), zookeeper.NoNodeException)
+        self.assertNotIn("Persistent retry error", output.getvalue())
+
+    @inlineCallbacks
+    def test_ephemeral_and_watch_recreate(self):
+        # Create some ephemeral nodes
+        yield self.client.create("/fo-1", "abc", flags=zookeeper.EPHEMERAL)
+        yield self.client.create("/fo-2", "def", flags=zookeeper.EPHEMERAL)
+
+        # Create some watches
+        g_d, g_w_d = self.client.get_and_watch("/fo-1")
+        yield g_d
+
+        c_d, c_w_d = self.client.get_children_and_watch("/")
+        yield g_d
+
+        e_d, e_w_d = self.client.get_children_and_watch("/fo-2")
+        yield e_d
+
+        # Expire the session
+        yield self.expire_session()
+
+        # Poof
+
+        # Ephemerals back
+        c, s = yield self.client.get("/fo-1")
+        self.assertEqual(c, "abc")
+
+        c, s = yield self.client.get("/fo-2")
+        self.assertEqual(c, "def")
+
+        # Watches triggered
+        yield DeferredList(
+            [g_w_d, c_w_d, e_w_d],
+            fireOnOneErrback=True, consumeErrors=True)
+
+        h = self.client.handle
+        self.assertEqual(
+            [str(d.result) for d in (g_w_d, c_w_d, e_w_d)],
+            ["<ClientEvent session at '/fo-1' state: connected handle:%d>" % h,
+             "<ClientEvent session at '/' state: connected handle:%d>" % h,
+             "<ClientEvent session at '/fo-2' state: connected handle:%d>" % h
+             ])
+
+    @inlineCallbacks
+    def test_ephemeral_no_track_sequence_nodes(self):
+        """ Ephemeral tracking ignores sequence nodes.
+        """
+        yield self.client.create("/music", "abc")
+        yield self.client.create(
+            "/music/u2-", "abc",
+            flags=zookeeper.EPHEMERAL | zookeeper.SEQUENCE)
+        yield self.expire_session()
+        children = yield self.client.get_children("/music")
+        self.assertEqual(children, [])
+
+    @inlineCallbacks
+    def test_ephemeral_content_modification(self):
+        yield self.client.create("/fo-1", "abc", flags=zookeeper.EPHEMERAL)
+        yield self.client.set("/fo-1", "def")
+        yield self.expire_session()
+        c, s = yield self.client.get("/fo-1")
+        self.assertEqual(c, "def")
+
+    @inlineCallbacks
+    def test_ephemeral_acl_modification(self):
+        yield self.client.create("/fo-1", "abc", flags=zookeeper.EPHEMERAL)
+        acl = [test_client.PUBLIC_ACL,
+               dict(scheme="digest",
+                    id="zebra:moon",
+                    perms=zookeeper.PERM_ALL)]
+        yield self.client.set_acl("/fo-1", acl)
+        yield self.expire_session()
+        n_acl, stat = yield self.client.get_acl("/fo-1")
+        self.assertEqual(acl, n_acl)
+
+    @inlineCallbacks
+    def test_ephemeral_deletion(self):
+        yield self.client.create("/fo-1", "abc", flags=zookeeper.EPHEMERAL)
+        yield self.client.delete("/fo-1")
+        yield self.expire_session()
+        self.assertFalse((yield self.client.exists("/fo-1")))

=== modified file 'txzookeeper/tests/test_node.py'
--- txzookeeper/tests/test_node.py	2011-06-30 17:14:06 +0000
+++ txzookeeper/tests/test_node.py	2013-05-15 13:35:05 +0000
@@ -1,8 +1,11 @@
 #
-#  Copyright (C) 2010 Canonical Ltd. All Rights Reserved
+#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
 #
 #  This file is part of txzookeeper.
 #
+#  Authors:
+#   Kapil Thangavelu
+#
 #  txzookeeper is free software: you can redistribute it and/or modify
 #  it under the terms of the GNU Lesser General Public License as published by
 #  the Free Software Foundation, either version 3 of the License, or
@@ -227,8 +230,8 @@
         """
         node = ZNode("/zoo/elephant", self.client)
         exists, watch = yield node.exists_and_watch()
-
-        self.client.create("/zoo/elephant")
+        self.assertFalse((yield exists))
+        yield self.client.create("/zoo/elephant")
         event = yield watch
         self.assertEqual(event.type, zookeeper.CREATED_EVENT)
         self.assertEqual(event.path, node.path)

=== modified file 'txzookeeper/tests/test_queue.py'
--- txzookeeper/tests/test_queue.py	2011-06-30 17:14:06 +0000
+++ txzookeeper/tests/test_queue.py	2013-05-15 13:35:05 +0000
@@ -1,8 +1,11 @@
 #
-#  Copyright (C) 2010 Canonical Ltd. All Rights Reserved
+#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
 #
 #  This file is part of txzookeeper.
 #
+#  Authors:
+#   Kapil Thangavelu
+#
 #  txzookeeper is free software: you can redistribute it and/or modify
 #  it under the terms of the GNU Lesser General Public License as published by
 #  the Free Software Foundation, either version 3 of the License, or
@@ -17,11 +20,11 @@
 #  along with txzookeeper.  If not, see <http://www.gnu.org/licenses/>.
 #
 
-
 from zookeeper import NoNodeException
 from twisted.internet.defer import (
     inlineCallbacks, returnValue, DeferredList, Deferred, succeed, fail)
 
+
 from txzookeeper import ZookeeperClient
 from txzookeeper.client import NotConnectedException
 from txzookeeper.queue import Queue, ReliableQueue, SerializedQueue, QueueItem

=== added file 'txzookeeper/tests/test_retry.py'
--- txzookeeper/tests/test_retry.py	1970-01-01 00:00:00 +0000
+++ txzookeeper/tests/test_retry.py	2013-05-15 13:35:05 +0000
@@ -0,0 +1,332 @@
+#
+#  Copyright (C) 2010-2012 Canonical Ltd. All Rights Reserved
+#
+#  This file is part of txzookeeper.
+#
+#  Authors:
+#   Kapil Thangavelu
+#
+#  txzookeeper is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU Lesser General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  txzookeeper is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public License
+#  along with txzookeeper.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+import json
+import time
+import zookeeper
+
+from twisted.internet.defer import inlineCallbacks, fail, succeed, Deferred
+
+from txzookeeper.client import ZookeeperClient
+from txzookeeper.retry import (
+    RetryClient, retry, retry_watch,
+    check_retryable, is_retryable, get_delay, sleep)
+
+from txzookeeper import retry as retry_module
+
+
+from txzookeeper.utils import retry_change
+
+from txzookeeper.tests import ZookeeperTestCase, utils
+from txzookeeper.tests.proxy import ProxyFactory
+from txzookeeper.tests import test_client
+
+
+class RetryCoreTests(ZookeeperTestCase):
+    """Test the retry functions in isolation.
+    """
+    timeout = 5
+
+    def test_is_retryable(self):
+        self.assertEqual(
+            is_retryable(zookeeper.SessionExpiredException()), False)
+        self.assertEqual(
+            is_retryable(zookeeper.ConnectionLossException()), True)
+        self.assertEqual(
+            is_retryable(zookeeper.OperationTimeoutException()), True)
+        self.assertEqual(
+            is_retryable(TypeError()), False)
+
+    def setup_always_retryable(self):
+        def check_retry(*args):
+            return True
+
+        self.patch(retry_module, "check_retryable", check_retry)
+
+    @inlineCallbacks
+    def test_sleep(self):
+        t = time.time()
+        yield sleep(0.5)
+        self.assertTrue(time.time() - t > 0.5)
+
+    @inlineCallbacks
+    def test_retry_function(self):
+        """The retry wrapper can be used for a function."""
+        self.setup_always_retryable()
+
+        results = [fail(zookeeper.ConnectionLossException()),
+                   fail(zookeeper.ConnectionLossException()),
+                   succeed(21)]
+
+        def original(zebra):
+            """Hello World"""
+            return results.pop(0)
+
+        result = yield retry(ZookeeperClient(), original, "magic")
+        self.assertEqual(result, 21)
+
+    @inlineCallbacks
+    def test_retry_method(self):
+        """The retry wrapper can be used for a method."""
+        self.setup_always_retryable()
+
+        results = [fail(zookeeper.ConnectionLossException()),
+                   fail(zookeeper.ConnectionLossException()),
+                   succeed(21)]
+
+        class Foobar(object):
+
+            def original(self, zebra):
+                """Hello World"""
+                return results.pop(0)
+
+        client = ZookeeperClient()
+        foo = Foobar()
+        result = yield retry(client, foo.original, "magic")
+        self.assertEqual(result, 21)
+
+    @inlineCallbacks
+    def test_retry_watch_function(self):
+        self.setup_always_retryable()
+
+        results = [(fail(zookeeper.ConnectionLossException()), Deferred()),
+                   (fail(zookeeper.ConnectionLossException()), Deferred()),
+                   (succeed(21), succeed(22))]
+
+        def original(zebra):
+            """Hello World"""
+            return results.pop(0)
+
+        client = ZookeeperClient()
+        value_d, watch_d = retry_watch(client, original, "magic")
+        self.assertEqual((yield value_d), 21)
+        self.assertEqual((yield watch_d), 22)
+
+    def test_check_retryable(self):
+        unrecoverable_errors = [
+            zookeeper.ApiErrorException(),
+            zookeeper.NoAuthException(),
+            zookeeper.NodeExistsException(),
+            zookeeper.SessionExpiredException(),
+            ]
+
+        class _Conn(object):
+            def __init__(self, **kw):
+                self.__dict__.update(kw)
+
+        error = zookeeper.ConnectionLossException()
+        conn = _Conn(connected=True, unrecoverable=False)
+        max_time = time.time() + 10
+
+        for e in unrecoverable_errors:
+            self.assertFalse(check_retryable(_Conn(), max_time, e))
+
+        self.assertTrue(check_retryable(conn, max_time, error))
+        self.assertFalse(check_retryable(conn, time.time() - 10, error))
+        self.assertFalse(check_retryable(
+            _Conn(connected=False, unrecoverable=False), max_time, error))
+        self.assertFalse(check_retryable(
+            _Conn(connected=True, unrecoverable=True), max_time, error))
+
+    def test_get_delay(self):
+        # Delay currently set to ~1/3 of session time.
+
+        # Verify max value is respected
+        self.assertEqual(get_delay(25000, 10), 7.5)
+        # Verify normal calculation
+        self.assertEqual(get_delay(15000, 10, 30), 4.5)
+
+
+class RetryClientTests(test_client.ClientTests):
+    """Run the full client test suite against the retry facade.
+    """
+    def setUp(self):
+        super(RetryClientTests, self).setUp()
+        self.client = RetryClient(ZookeeperClient("127.0.0.1:2181", 3000))
+        self.client2 = None
+
+    def tearDown(self):
+        if self.client.connected:
+            utils.deleteTree(handle=self.client.handle)
+            self.client.close()
+
+        if self.client2 and self.client2.connected:
+            self.client2.close()
+
+        super(RetryClientTests, self).tearDown()
+
+    def test_wb_connect_after_timeout(self):
+        """white box tests disabled for retryclient."""
+
+    def test_wb_reconnect_after_timeout_and_close(self):
+        """white box tests disabled for retryclient."""
+
+    def test_exists_with_error(self):
+        """White box tests disabled for retryclient."""
+
+
+class RetryClientConnectionLossTest(ZookeeperTestCase):
+
+    def setUp(self):
+        super(RetryClientConnectionLossTest, self).setUp()
+
+        from twisted.internet import reactor
+        self.proxy = ProxyFactory("127.0.0.1", 2181)
+        self.proxy_port = reactor.listenTCP(0, self.proxy)
+        host = self.proxy_port.getHost()
+        self.proxied_client = RetryClient(ZookeeperClient(
+            "%s:%s" % (host.host, host.port)))
+        self.direct_client = ZookeeperClient("127.0.0.1:2181", 3000)
+        self.session_events = []
+
+        def session_event_collector(conn, event):
+            self.session_events.append(event)
+
+        self.proxied_client.set_session_callback(session_event_collector)
+        return self.direct_client.connect()
+
+    @inlineCallbacks
+    def tearDown(self):
+        import zookeeper
+        zookeeper.set_debug_level(0)
+        if self.proxied_client.connected:
+            yield self.proxied_client.close()
+        if not self.direct_client.connected:
+            yield self.direct_client.connect()
+        utils.deleteTree(handle=self.direct_client.handle)
+        yield self.direct_client.close()
+        self.proxy.lose_connection()
+        yield self.proxy_port.stopListening()
+
+    @inlineCallbacks
+    def test_get_children_and_watch(self):
+        yield self.proxied_client.connect()
+
+        # Setup tree
+        cpath = "/test-tree"
+        yield self.direct_client.create(cpath)
+
+        # Block the request (drops all packets.)
+        self.proxy.set_blocked(True)
+        child_d, watch_d = self.proxied_client.get_children_and_watch(cpath)
+
+        # Unblock and disconnect
+        self.proxy.set_blocked(False)
+        self.proxy.lose_connection()
+
+        # Call goes through
+        self.assertEqual((yield child_d), [])
+        self.assertEqual(len(self.session_events), 2)
+
+        # And we have reconnect events
+        self.assertEqual(self.session_events[-1].state_name, "connected")
+
+        yield self.direct_client.create(cpath + "/abc")
+
+        # The original watch is still active
+        yield watch_d
+
+    @inlineCallbacks
+    def test_exists_and_watch(self):
+        yield self.proxied_client.connect()
+
+        cpath = "/test-tree"
+
+        # Block the request
+        self.proxy.set_blocked(True)
+        exists_d, watch_d = self.proxied_client.exists_and_watch(cpath)
+
+        # Create the node
+        yield self.direct_client.create(cpath)
+
+        # Unblock and disconnect
+        self.proxy.set_blocked(False)
+        self.proxy.lose_connection()
+
+        # Call gets retried, see the latest state
+        self.assertTrue((yield exists_d))
+        self.assertEqual(len(self.session_events), 2)
+
+        # And we have reconnect events
+        self.assertEqual(self.session_events[-1].state_name, "connected")
+
+        yield self.direct_client.delete(cpath)
+
+        # The original watch is still active
+        yield watch_d
+
+    @inlineCallbacks
+    def test_get_and_watch(self):
+        yield self.proxied_client.connect()
+
+        # Setup tree
+        cpath = "/test-tree"
+        yield self.direct_client.create(cpath)
+
+        # Block the request (drops all packets.)
+        self.proxy.set_blocked(True)
+        get_d, watch_d = self.proxied_client.get_and_watch(cpath)
+
+        # Unblock and disconnect
+        self.proxy.set_blocked(False)
+        self.proxy.lose_connection()
+
+        # Call goes through
+        content, stat = yield get_d
+        self.assertEqual(content, '')
+        self.assertEqual(len(self.session_events), 2)
+
+        # And we have reconnect events
+        self.assertEqual(self.session_events[-1].state_name, "connected")
+
+        yield self.direct_client.delete(cpath)
+
+        # The original watch is still active
+        yield watch_d
+
+    @inlineCallbacks
+    def test_set(self):
+        yield self.proxied_client.connect()
+
+        # Setup tree
+        cpath = "/test-tree"
+        yield self.direct_client.create(cpath, json.dumps({"a": 1, "c": 2}))
+
+        def update_node(content, stat):
+            data = json.loads(content)
+            data["a"] += 1
+            data["b"] = 0
+            return json.dumps(data)
+
+        # Block the request (drops all packets.)
+        self.proxy.set_blocked(True)
+        mod_d = retry_change(self.proxied_client, cpath, update_node)
+
+        # Unblock and disconnect
+        self.proxy.set_blocked(False)
+        self.proxy.lose_connection()
+
+        # Call goes through, contents verified.
+        yield mod_d
+        content, stat = yield self.direct_client.get(cpath)
+        self.assertEqual(json.loads(content),
+                         {"a": 2, "b": 0, "c": 2})

=== modified file 'txzookeeper/tests/test_security.py'
--- txzookeeper/tests/test_security.py	2011-06-30 17:14:06 +0000
+++ txzookeeper/tests/test_security.py	2013-05-15 13:35:05 +0000
@@ -1,8 +1,11 @@
 #
-#  Copyright (C) 2010 Canonical Ltd. All Rights Reserved
+#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
 #
 #  This file is part of txzookeeper.
 #
+#  Authors:
+#   Kapil Thangavelu
+#
 #  txzookeeper is free software: you can redistribute it and/or modify
 #  it under the terms of the GNU Lesser General Public License as published by
 #  the Free Software Foundation, either version 3 of the License, or

=== modified file 'txzookeeper/tests/test_session.py'
--- txzookeeper/tests/test_session.py	2011-06-30 20:16:17 +0000
+++ txzookeeper/tests/test_session.py	2013-05-15 13:35:05 +0000
@@ -1,21 +1,46 @@
+#
+#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
+#
+#  This file is part of txzookeeper.
+#
+#  Authors:
+#   Kapil Thangavelu
+#
+#  txzookeeper is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU Lesser General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  txzookeeper is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public License
+#  along with txzookeeper.  If not, see <http://www.gnu.org/licenses/>.
+#
 
 import atexit
+import logging
 import os
 
 import zookeeper
 
-from twisted.internet import reactor
-from twisted.internet.defer import inlineCallbacks, Deferred, returnValue
+from twisted.internet.defer import (
+    inlineCallbacks, Deferred, DeferredList, returnValue)
 
 from txzookeeper import ZookeeperClient
 from txzookeeper.client import NotConnectedException, ConnectionException
 
 from txzookeeper.tests.common import ZookeeperCluster
 from txzookeeper.tests import ZookeeperTestCase
+from txzookeeper import managed
 
 
 ZK_HOME = os.environ.get("ZOOKEEPER_PATH")
-assert ZK_HOME, "ZOOKEEPER_PATH environment variable must be defined"
+assert ZK_HOME, (
+    "ZOOKEEPER_PATH environment variable must be defined.\n "
+    "For deb package installations this is /usr/share/java")
 
 CLUSTER = ZookeeperCluster(ZK_HOME)
 atexit.register(lambda cluster: cluster.terminate(), CLUSTER)
@@ -31,18 +56,16 @@
         zookeeper.deterministic_conn_order(True)
         zookeeper.set_debug_level(0)
 
-    def sleep(self, delay):
-        """Non-blocking sleep."""
-        deferred = Deferred()
-        reactor.callLater(delay, deferred.callback, None)
-        return deferred
-
     @property
     def cluster(self):
         return CLUSTER
 
     def tearDown(self):
         super(ClientSessionTests, self).tearDown()
+        if self.client:
+            self.client.close()
+        if self.client2:
+            self.client2.close()
         self.cluster.reset()
 
     @inlineCallbacks
@@ -118,11 +141,18 @@
         """A callback can be specified for connection errors.
 
         We can specify a callback for connection errors, that
-        can perform recovery for a disconnected client, restablishing
+        can perform recovery for a disconnected client.
         """
         @inlineCallbacks
         def connection_error_handler(connection, error):
+            # Moved management of this connection attribute out of the
+            # default behavior for a connection exception, to support
+            # the retry facade. Under the hood libzk is going to be
+            # trying to transparently reconnect
+            connection.connected = False
+
             # On loss of the connection, reconnect the client w/ same session.
+            connection.close()
             yield connection.connect(
                 self.cluster[1].address, client_id=connection.client_id)
             returnValue(23)
@@ -157,7 +187,7 @@
 
         def session_event_callback(connection, e):
             session_events.append(e)
-            if len(session_events) == 4:
+            if len(session_events) == 8:
                 events_received.callback(True)
 
         # Connect to a node in the cluster and establish a watch
@@ -165,11 +195,18 @@
         self.client.set_session_callback(session_event_callback)
         yield self.client.connect()
 
-        child_d, watch_d = self.client.get_children_and_watch("/")
-        yield child_d
+        # Setup some watches to verify they are cleaned out on expiration.
+        d, e_watch_d = self.client.exists_and_watch("/")
+        yield d
+
+        d, g_watch_d = self.client.get_and_watch("/")
+        yield d
+
+        d, c_watch_d = self.client.get_children_and_watch("/")
+        yield d
 
         # Connect a client to the same session on a different node.
-        self.client2 = ZookeeperClient(self.cluster[0].address)
+        self.client2 = ZookeeperClient(self.cluster[1].address)
         yield self.client2.connect(client_id=self.client.client_id)
 
         # Close the new client and wait for the event propogation
@@ -177,8 +214,11 @@
 
         # It can take some time for this to propagate
         yield events_received
-        self.assertEqual(len(session_events), 4)
-        self.assertEqual(session_events[-1].state_name, "expired")
+        self.assertEqual(len(session_events), 8)
+
+        # The last four (conn + 3 watches) are all expired
+        for evt in session_events[4:]:
+            self.assertEqual(evt.state_name, "expired")
 
         # The connection is dead without reconnecting.
         yield self.assertFailure(
@@ -186,9 +226,17 @@
             NotConnectedException, ConnectionException)
 
         self.assertTrue(self.client.unrecoverable)
+        yield self.assertFailure(e_watch_d, zookeeper.SessionExpiredException)
+        yield self.assertFailure(g_watch_d, zookeeper.SessionExpiredException)
+        yield self.assertFailure(c_watch_d, zookeeper.SessionExpiredException)
 
         # If a reconnect attempt is made with a dead session id
-        #yield self.client.connect(client_id=self.client.client_id)
+        client_id = self.client.client_id
+        self.client.close()  # Free the handle
+        yield self.client.connect(client_id=client_id)
+        yield self.assertFailure(
+            self.client.get_children("/"),
+            NotConnectedException, ConnectionException)
 
     test_client_session_expiration_event.timeout = 10
 
@@ -210,13 +258,13 @@
             session_events.append(e)
 
         # Connect to a node in the cluster and establish a watch
-        self.client = ZookeeperClient(self.cluster[2].address)
+        self.client = ZookeeperClient(self.cluster[2].address,
+                                      session_timeout=5000)
         self.client.set_session_callback(session_event_callback)
         yield self.client.connect()
 
         yield self.client.create("/hello", flags=zookeeper.EPHEMERAL)
-        exists_d, watch_d = self.client.exists_and_watch("/hello")
-        yield exists_d
+        self.assertTrue((yield self.client.exists("/hello")))
 
         # Shutdown the server the client is connected to
         self.cluster[2].stop()
@@ -243,3 +291,46 @@
         # Ephemeral is destroyed when the session closed.
         exists = yield self.client.exists("/hello")
         self.assertFalse(exists)
+
+    @inlineCallbacks
+    def test_managed_client_backoff(self):
+        output = self.capture_log(level=logging.DEBUG)
+        self.patch(managed, 'BACKOFF_INCREMENT', 2)
+        self.client = yield managed.ManagedClient(
+            self.cluster[0].address,
+            connect_timeout=4).connect()
+        self.client2 = yield ZookeeperClient(self.cluster[1].address).connect()
+
+        exists_d, watch_d = self.client.exists_and_watch("/hello")
+        yield exists_d
+
+        yield self.client2.create("/hello", "world")
+        yield self.client2.close()
+
+        self.cluster[0].stop()
+        yield self.sleep(1)
+
+        # Try to do something with the connection while its down.
+        ops = []
+        ops.append(self.client.create('/abc', 'test'))
+        ops.append(self.client.get("/hello"))
+        ops.append(self.client.get_children("/"))
+        ops.append(self.client.set("/hello", "sad"))
+
+        # Sleep and let the session expire, and ensure we're down long enough
+        # for backoff to trigger.
+        yield self.sleep(10)
+
+        # Start the cluster and watch things work
+        self.cluster[0].run()
+        yield DeferredList(
+            ops, fireOnOneErrback=True, consumeErrors=True)
+
+        yield watch_d
+
+        # Verify we backed off at least once
+        self.assertIn("Backing off reconnect", output.getvalue())
+        # Verify we only reconnected once
+        self.assertTrue(output.getvalue().count("Restablished connection"), 1)
+
+    test_managed_client_backoff.timeout = 25

=== modified file 'txzookeeper/tests/test_utils.py'
--- txzookeeper/tests/test_utils.py	2011-06-30 17:14:06 +0000
+++ txzookeeper/tests/test_utils.py	2013-05-15 13:35:05 +0000
@@ -1,8 +1,11 @@
 #
-#  Copyright (C) 2010 Canonical Ltd. All Rights Reserved
+#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
 #
 #  This file is part of txzookeeper.
 #
+#  Authors:
+#   Kapil Thangavelu
+#
 #  txzookeeper is free software: you can redistribute it and/or modify
 #  it under the terms of the GNU Lesser General Public License as published by
 #  the Free Software Foundation, either version 3 of the License, or

=== modified file 'txzookeeper/tests/utils.py'
--- txzookeeper/tests/utils.py	2011-06-30 17:14:06 +0000
+++ txzookeeper/tests/utils.py	2013-05-15 13:35:05 +0000
@@ -1,8 +1,11 @@
 #
-#  Copyright (C) 2010 Canonical Ltd. All Rights Reserved
+#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
 #
 #  This file is part of txzookeeper.
 #
+#  Authors:
+#   Kapil Thangavelu
+#
 #  txzookeeper is free software: you can redistribute it and/or modify
 #  it under the terms of the GNU Lesser General Public License as published by
 #  the Free Software Foundation, either version 3 of the License, or

=== modified file 'txzookeeper/todo.txt'
--- txzookeeper/todo.txt	2010-07-12 07:42:22 +0000
+++ txzookeeper/todo.txt	2013-05-15 13:35:05 +0000
@@ -1,10 +1,2 @@
 
-bugs to file upstream
-
- - you can set acl on a non existant node.
- - memory leak every api invocation. [really? need some measurements here]
-
- observed while trying xtest_get_children_with_watcher
-
- - segfault if close during completion.
- - getting a watch notification when closing a connection, segfaults.
+Documentation

=== modified file 'txzookeeper/utils.py'
--- txzookeeper/utils.py	2011-06-17 23:52:34 +0000
+++ txzookeeper/utils.py	2013-05-15 13:35:05 +0000
@@ -1,8 +1,11 @@
 #
-#  Copyright (C) 2010 Canonical Ltd. All Rights Reserved
+#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
 #
 #  This file is part of txzookeeper.
 #
+#  Authors:
+#   Kapil Thangavelu
+#
 #  txzookeeper is free software: you can redistribute it and/or modify
 #  it under the terms of the GNU Lesser General Public License as published by
 #  the Free Software Foundation, either version 3 of the License, or
@@ -17,8 +20,9 @@
 #  along with txzookeeper.  If not, see <http://www.gnu.org/licenses/>.
 #
 
+
 import zookeeper
-from twisted.internet.defer import inlineCallbacks
+from twisted.internet.defer import inlineCallbacks, Deferred
 
 
 @inlineCallbacks
@@ -65,3 +69,16 @@
                 zookeeper.NoNodeException,
                 zookeeper.BadVersionException):
             pass
+
+
+def sleep(delay):
+    """Non-blocking sleep.
+
+    :param int delay: time in seconds to sleep.
+    :return: a Deferred that fires after the desired delay.
+    :rtype: :class:`twisted.internet.defer.Deferred`
+    """
+    from twisted.internet import reactor
+    deferred = Deferred()
+    reactor.callLater(delay, deferred.callback, None)
+    return deferred