Merge lp:~hazmat/txzookeeper/managed-watch-and-ephemeral into lp:txzookeeper

[Kapil Thangavelu]

Managed Connections for txzk

A managed connection automatically handles transient connection failures,
and session expiration. On session expiration it recreates ephemeral nodes,
and fire watch events on extant watchers.

https://codereview.appspot.com/5976074/

[Kapil Thangavelu (hazmat)]

Reviewers: mp+100715_code.launchpad.net,

Message:
Please take a look.

Description:
Managed Connections for txzk

A managed connection automatically handles transient connection
failures,
and session expiration. On session expiration it recreates ephemeral
nodes,
and fire watch events on extant watchers.

https://code.launchpad.net/~hazmat/txzookeeper/managed-watch-and-ephemeral/+merge/100715

(do not edit description out of merge proposal)

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

Affected files:
   A .bzrignore
   A [revision details]
   M setup.py
   M txzookeeper/client.py
   A txzookeeper/managed.py
   M txzookeeper/retry.py
   M txzookeeper/tests/__init__.py
   M txzookeeper/tests/test_client.py
   A txzookeeper/tests/test_managed.py
   M txzookeeper/tests/test_retry.py

[Kapil Thangavelu (hazmat)]

Reviewers: mp+100715_code.launchpad.net,

Message:
Please take a look.

Description:
Managed Connections for txzk

A managed connection automatically handles transient connection
failures,
and session expiration. On session expiration it recreates ephemeral
nodes,
and fire watch events on extant watchers.

https://code.launchpad.net/~hazmat/txzookeeper/managed-watch-and-ephemeral/+merge/100715

(do not edit description out of merge proposal)

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

Affected files:
   A .bzrignore
   A [revision details]
   M setup.py
   M txzookeeper/client.py
   A txzookeeper/managed.py
   M txzookeeper/retry.py
   M txzookeeper/tests/__init__.py
   M txzookeeper/tests/test_client.py
   A txzookeeper/tests/test_managed.py
   M txzookeeper/tests/test_retry.py

[lp:~hazmat/txzookeeper/managed-watch-and-ephemeral]

55. By Kapil Thangavelu on 2012-04-04

remove unesc. expire handler from retry client

[William Reade (fwereade)]

This looks very nice indeed, but the vagueness of the "comes at a cost"
comment makes me wonder what I'm missing...

https://codereview.appspot.com/5976074/diff/1/txzookeeper/managed.py
File txzookeeper/managed.py (right):

https://codereview.appspot.com/5976074/diff/1/txzookeeper/managed.py#newcode66
txzookeeper/managed.py:66: mgr # keep the flakes happy
Can't you just do:

   with self._ctx():

?

https://codereview.appspot.com/5976074/diff/1/txzookeeper/managed.py#newcode70
txzookeeper/managed.py:70: mgr # keep the flakes happy
Ditto

https://codereview.appspot.com/5976074/diff/1/txzookeeper/managed.py#newcode122
txzookeeper/managed.py:122: come at a cost though.
Please expand ;).

https://codereview.appspot.com/5976074/diff/1/txzookeeper/managed.py#newcode138
txzookeeper/managed.py:138: """Called on intercept of session expiration
to restablish the session.
I'm pretty sure there are 2 es in re-establish, and that the hyphen may
have fallen out of common use but remains IMO more readable. As you
wish.

https://codereview.appspot.com/5976074/diff/1/txzookeeper/managed.py#newcode140
txzookeeper/managed.py:140: This will reconnect to zk, restabslish
ephemerals, and trigger watches.
I like the word "restabslish", but ditto.

https://codereview.appspot.com/5976074/diff/1/txzookeeper/tests/test_managed.py
File txzookeeper/tests/test_managed.py (right):

https://codereview.appspot.com/5976074/diff/1/txzookeeper/tests/test_managed.py#newcode169
txzookeeper/tests/test_managed.py:169: # It takes some time to propogate
(1/3 session time as ping)
s/propogate/propagate/

https://codereview.appspot.com/5976074/

[Kapil Thangavelu (hazmat)]

On 2012/04/04 13:23:29, fwereade wrote:
> This looks very nice indeed, but the vagueness of the "comes at a
cost" comment
> makes me wonder what I'm missing...

expanded greatly what those costs are.

> https://codereview.appspot.com/5976074/diff/1/txzookeeper/managed.py
> File txzookeeper/managed.py (right):

https://codereview.appspot.com/5976074/diff/1/txzookeeper/managed.py#newcode66
> txzookeeper/managed.py:66: mgr # keep the flakes happy
> Can't you just do:

> with self._ctx():

> ?

much nicer, thanks.

https://codereview.appspot.com/5976074/diff/1/txzookeeper/managed.py#newcode70
> txzookeeper/managed.py:70: mgr # keep the flakes happy
> Ditto

done.

https://codereview.appspot.com/5976074/diff/1/txzookeeper/managed.py#newcode122
> txzookeeper/managed.py:122: come at a cost though.
> Please expand ;).

done.

https://codereview.appspot.com/5976074/diff/1/txzookeeper/managed.py#newcode138
> txzookeeper/managed.py:138: """Called on intercept of session
expiration to
> restablish the session.
> I'm pretty sure there are 2 es in re-establish, and that the hyphen
may have
> fallen out of common use but remains IMO more readable. As you wish.

proper speling established ;-)

https://codereview.appspot.com/5976074/diff/1/txzookeeper/managed.py#newcode140
> txzookeeper/managed.py:140: This will reconnect to zk, restabslish
ephemerals,
> and trigger watches.
> I like the word "restabslish", but ditto.

https://codereview.appspot.com/5976074/diff/1/txzookeeper/tests/test_managed.py
> File txzookeeper/tests/test_managed.py (right):

https://codereview.appspot.com/5976074/diff/1/txzookeeper/tests/test_managed.py#newcode169
> txzookeeper/tests/test_managed.py:169: # It takes some time to
propogate (1/3
> session time as ping)
> s/propogate/propagate/

done.

http://codereview.appspot.com/5976074/

[Kapil Thangavelu (hazmat)]

Please take a look.

https://codereview.appspot.com/5976074/diff/1/txzookeeper/managed.py
File txzookeeper/managed.py (right):

https://codereview.appspot.com/5976074/diff/1/txzookeeper/managed.py#newcode66
txzookeeper/managed.py:66: mgr # keep the flakes happy
On 2012/04/04 13:23:29, fwereade wrote:
> Can't you just do:

> with self._ctx():

> ?

thanks, much nicer.

https://codereview.appspot.com/5976074/

[lp:~hazmat/txzookeeper/managed-watch-and-ephemeral]

56. By Kapil Thangavelu on 2012-04-04

address review comments

[Jim Baker (jimbaker)]

+1, LGTM, just some minors

https://codereview.appspot.com/5976074/diff/4001/txzookeeper/managed.py
File txzookeeper/managed.py (right):

https://codereview.appspot.com/5976074/diff/4001/txzookeeper/managed.py#newcode153
txzookeeper/managed.py:153: def cb_restablish_session(self, e=None):
Presumably this should be cb_re_establish... but I don't see the
difference here with the easier synonym: restore. Good either way with
me.

https://codereview.appspot.com/5976074/diff/4001/txzookeeper/managed.py#newcode162
txzookeeper/managed.py:162: # If its been explicitly closed, don't
restablish.
it's... re-establish

(or are you saying something about being restful? ;) )

https://codereview.appspot.com/5976074/diff/4001/txzookeeper/managed.py#newcode166
txzookeeper/managed.py:166: # If its a stale handle, don't restablish
it's... re-establish

https://codereview.appspot.com/5976074/diff/4001/txzookeeper/managed.py#newcode174
txzookeeper/managed.py:174: # Its already been restablished, don't
restablish.
It's... re-established... re-establish

https://codereview.appspot.com/5976074/diff/4001/txzookeeper/managed.py#newcode183
txzookeeper/managed.py:183: # Restablish
I'm not going to mark any more of these!

https://codereview.appspot.com/5976074/diff/4001/txzookeeper/managed.py#newcode189
txzookeeper/managed.py:189: log.error("error while restablish %r %s" %
(e, e))
Another minor: probably want some consistency in beginning with
uppercase. Not certain why you use log.error here, but log.exception
later for what appears to be a similar case.

https://codereview.appspot.com/5976074/diff/4001/txzookeeper/tests/test_managed.py
File txzookeeper/tests/test_managed.py (right):

https://codereview.appspot.com/5976074/diff/4001/txzookeeper/tests/test_managed.py#newcode110
txzookeeper/tests/test_managed.py:110: """Reset fires a synthentic
client event, and clears watches.
no need for comma

https://codereview.appspot.com/5976074/

=== added file '.bzrignore'
--- .bzrignore	1970-01-01 00:00:00 +0000
+++ .bzrignore	2012-04-04 18:12:22 +0000
@@ -0,0 +1,4 @@
+.emacs.desktop
+_trial_temp
+txzookeeper
+txzookeeper.egg-info
 
=== modified file 'setup.py'
--- setup.py	2011-12-12 13:19:28 +0000
+++ setup.py	2012-04-04 18:12:22 +0000
@@ -45,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(),
 
=== modified file 'txzookeeper/client.py'
--- txzookeeper/client.py	2011-12-06 21:21:29 +0000
+++ txzookeeper/client.py	2012-04-04 18:12:22 +0000
@@ -96,15 +96,20 @@
 
     @property
     def state_name(self):
-        return STATE_NAME_MAPPING[self.args[2]]
+        return STATE_NAME_MAPPING.get(self.args[2], "unknown")
 
     @property
     def type_name(self):
-        return TYPE_NAME_MAPPING[self.args[1]]
+        return TYPE_NAME_MAPPING.get(self.args[1], "unknown")
+
+    @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):
@@ -133,11 +138,11 @@
 
     @property
     def type_name(self):
-        return TYPE_NAME_MAPPING[self.type]
+        return TYPE_NAME_MAPPING.get(self.type, "unknown")
 
     @property
     def state_name(self):
-        return STATE_NAME_MAPPING[self.connection_state]
+        return STATE_NAME_MAPPING.get(self.connection_state, "unknown")
 
     def __repr__(self):
         return  "<ClientEvent %s at %r state: %s>" % (
@@ -169,6 +174,10 @@
         self.connected = False
         self.handle = None
 
+    def __repr__(self):
+        return  "<txZookeeperClient client: %s handle: %r state: %s>" % (
+            self.client_id[0], self.handle, self.state)
+
     def _check_connected(self, d):
         if not self.connected:
             d.errback(NotConnectedException("not connected"))
@@ -218,7 +227,7 @@
             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)
         return d
@@ -234,7 +243,7 @@
             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)
         return d
@@ -251,12 +260,12 @@
             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)
         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):
@@ -285,14 +294,19 @@
         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
@@ -391,6 +405,7 @@
         d = defer.Deferred()
         if self._check_result(result, d):
             return d
+        self.handle = None
         d.callback(True)
         return d
 
@@ -437,7 +452,6 @@
         else:
             self.handle = zookeeper.init(
                 self._servers, callback, self._session_timeout)
-
         return d
 
     def _cb_connected(
@@ -460,6 +474,7 @@
             # If we timed out and then connected, then close the conn.
             if state == zookeeper.CONNECTED_STATE and scheduled_timeout.called:
                 self.close()
+                return
 
             # Send session events to the callback, in addition to any
             # duplicate session events that will be sent for extant watches.
@@ -468,7 +483,10 @@
                     self, ClientEvent(type, state, path))
 
             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
 
@@ -488,17 +506,18 @@
         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)
         return d
 
+    def _cb_created(self, d, data, acls, flags, result_code, path):
+        if self._check_result(result_code, d):
+            return
+        d.callback(path)
+
     def delete(self, path, version=-1):
         """
         Delete the node at the given path. If the current node version on the
@@ -513,16 +532,16 @@
         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)
         return d
 
+    def _cb_deleted(self, d, path, result_code):
+        if self._check_result(result_code, d):
+            return
+        d.callback(result_code)
+
     def exists(self, path):
         """
         Check that the given node path exists. Returns a deferred that
@@ -657,17 +676,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)
         return d
 
+    def _cb_set_acl(self, d, path, acls, result_code):
+        if self._check_result(result_code, d):
+            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
@@ -683,16 +702,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)
         return d
 
+    def _cb_set(self, d, path, data, result_code, node_stat):
+        if self._check_result(result_code, d):
+            return
+        d.callback(node_stat)
+
     def set_connection_watcher(self, watcher):
         """
         Sets a permanent global watcher on the connection. This will get
@@ -702,7 +721,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):
@@ -740,9 +759,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.
 
=== added file 'txzookeeper/managed.py'
--- txzookeeper/managed.py	1970-01-01 00:00:00 +0000
+++ txzookeeper/managed.py	2012-04-04 18:12:22 +0000
@@ -0,0 +1,350 @@
+from functools import partial
+
+import contextlib
+import logging
+import zookeeper
+
+from twisted.internet.defer import (
+    inlineCallbacks, DeferredLock, fail, returnValue)
+
+from client import ZookeeperClient, ClientEvent, NotConnectedException
+from retry import RetryClient
+
+
+class StopWatcher(Exception):
+    pass
+
+
+WATCH_KIND_MAP = {
+    "child": "get_children_and_watch",
+    "exists": "exists_and_watch",
+    "get": "get_and_watch"
+    }
+
+
+log = logging.getLogger("txzk.managed")
+
+
+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.
+    """
+
+    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._reconnect_lock = DeferredLock()
+        self.set_connection_error_callback(self._cb_connection_error)
+
+    @inlineCallbacks
+    def cb_restablish_session(self, e=None):
+        """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()
+
+        try:
+            # If its been explicitly closed, don't restablish.
+            if self.handle is None:
+                return
+
+            # If its a stale handle, don't restablish
+            try:
+                zookeeper.is_unrecoverable(self.handle)
+            except zookeeper.ZooKeeperException:
+                if e:
+                    raise e
+                return
+
+            # Its already been restablished, don't restablish.
+            if not self.unrecoverable:
+                return
+            elif self.connected:
+                self.close()
+                self.handle = 0
+            elif isinstance(self.handle, int):
+                self.handle = 0
+
+            # Restablish
+            assert self.handle == 0
+            yield self._cb_restablish_session().addErrback(
+                self._cb_restablish_errback, e)
+
+        except Exception, e:
+            log.error("error while restablish %r %s" % (e, e))
+        finally:
+            yield self._reconnect_lock.release()
+
+    @inlineCallbacks
+    def _cb_restablish_session(self):
+        """Re-establish a new session, and recreate ephemerals and watches.
+        """
+        while 1:
+            # Reconnect
+            if self.handle is None:
+                returnValue(self.handle)
+            try:
+                yield self.connect(timeout=self._connect_timeout)
+            except zookeeper.ZooKeeperException, e:
+                log.exception("Error while connecting %r %s" % (e, e))
+                continue
+            else:
+                break
+
+        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)
+        yield self._watches.reset()
+
+    def _cb_restablish_errback(self, err, failure):
+        """If there's an error re-establishing the session log it.
+        """
+        log.error("Error while trying to restablish connection %s\n%s" % (
+            failure.value, failure.getTraceback()))
+        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 isinstance(error, (
+            zookeeper.SessionExpiredException,
+            NotConnectedException,
+            zookeeper.ClosingException)):
+            raise error
+        yield self._cb_restablish_session()
+        raise zookeeper.ConnectionLossException
+
+    @inlineCallbacks
+    def _cb_connection_event(self, evt):
+        """
+        """
+
+    # client connected tracker
+    def _check_connected(self, d):
+        """Clients are automatically reconnected."""
+        if self.connected:
+            return
+
+        if self.handle is None:
+            d.errback(NotConnectedException("not connected"))
+            return d
+
+        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.
+            """
+            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):
+            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))
+        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):
+            return
+
+        if flags & zookeeper.EPHEMERAL:
+            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):
+            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):
+            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):
+            return
+
+        if path in self._ephemerals:
+            self._ephemerals[path]['data'] = data
+
+        d.callback(node_stat)
+
+
+class ManagedClient(RetryClient):
+
+    def __init__(self, *args, **kw):
+        client = SessionClient(*args, **kw)
+        super(ManagedClient, self).__init__(
+            client, client.cb_restablish_session)
+
+
+def ManagedClient(servers=None, session_timeout=None, connect_timeout=10000):
+    from retry import RetryClient
+    client = SessionClient(servers, session_timeout, connect_timeout)
+    return RetryClient(client)
 
=== modified file 'txzookeeper/retry.py'
--- txzookeeper/retry.py	2011-11-08 12:15:16 +0000
+++ txzookeeper/retry.py	2012-04-04 18:12:22 +0000
@@ -64,7 +64,7 @@
 
     :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 fractinoal amount of a timeout to wait
+    :param session_fraction: The fractional amount of a timeout to wait
     """
     retry_delay = session_timeout / (float(session_fraction) * 1000)
     return min(retry_delay, max_delay)
@@ -111,20 +111,20 @@
            must return a single value (either a deferred or result
            value).
     """
-    # For clients which aren't connected (session timeout == None)
-    # we raise the errors to the callers
-    session_timeout = client.session_timeout or 0
-
-    # If we keep retrying past the 1.5 * session timeout without
-    # success just die, the session expiry is fatal.
-    max_time = session_timeout * 1.5 + time.time()
-
     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
+
+            # If we keep retrying past the 1.5 * session timeout without
+            # success just die, the session expiry is fatal.
+            max_time = session_timeout * 1.5 + time.time()
             if not check_retryable(client, max_time, e):
                 raise
+
             # Give the connection a chance to auto-heal.
             yield sleep(get_delay(session_timeout))
             continue
@@ -172,6 +172,7 @@
         # Give the connection a chance to auto-heal
         d = sleep(get_delay(session_timeout))
         d.addCallback(retry_inner)
+
         return d
 
     def retry_inner(value):
 
=== modified file 'txzookeeper/tests/__init__.py'
--- txzookeeper/tests/__init__.py	2011-10-22 04:24:02 +0000
+++ txzookeeper/tests/__init__.py	2012-04-04 18:12:22 +0000
@@ -20,6 +20,8 @@
 #  along with txzookeeper.  If not, see <http://www.gnu.org/licenses/>.
 #
 
+import logging
+import StringIO
 import sys
 
 import zookeeper
@@ -72,6 +74,26 @@
             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/test_client.py'
--- txzookeeper/tests/test_client.py	2011-10-22 04:24:02 +0000
+++ txzookeeper/tests/test_client.py	2012-04-04 18:12:22 +0000
@@ -1050,7 +1050,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
 
=== added file 'txzookeeper/tests/test_managed.py'
--- txzookeeper/tests/test_managed.py	1970-01-01 00:00:00 +0000
+++ txzookeeper/tests/test_managed.py	2012-04-04 18:12:22 +0000
@@ -0,0 +1,263 @@
+#
+#  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 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()
+        self.assertEqual(
+            str(ClientEvent(*e)),
+            "<ClientEvent session at '/foobar' state: connected>")
+        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()
+        self.assertEqual(
+            str(ClientEvent(*e)),
+            "<ClientEvent session at '/foobar' state: connected>")
+        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
+        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):
+        assert self.client.connected
+        self.client2 = ZookeeperClient(self.client.servers)
+        yield self.client2.connect(client_id=self.client.client_id)
+        yield self.client2.close()
+        # It takes some time to propagate (1/3 session time as ping)
+        yield self.sleep(2)
+
+    @inlineCallbacks
+    def test_session_expiration_conn(self):
+        session_id = self.client.client_id[0]
+        yield self.client.create("/fo-1", "abc")
+        yield self.expire_session()
+        yield self.client.exists("/")
+        self.assertNotEqual(session_id, self.client.client_id[0])
+
+    @inlineCallbacks
+    def test_session_expiration_conn_watch(self):
+        session_id = self.client.client_id[0]
+        yield self.client.create("/fo-1", "abc")
+        yield self.expire_session()
+        yield self.client.exists("/")
+        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_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)
+
+        self.assertEqual(
+            [str(d.result) for d in (g_w_d, c_w_d, e_w_d)],
+            ["<ClientEvent session at '/fo-1' state: connected>",
+             "<ClientEvent session at '/' state: connected>",
+             "<ClientEvent session at '/fo-2' state: connected>"])
+
+    @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_retry.py'
--- txzookeeper/tests/test_retry.py	2011-11-08 12:15:16 +0000
+++ txzookeeper/tests/test_retry.py	2012-04-04 18:12:22 +0000
@@ -1,5 +1,5 @@
 #
-#  Copyright (C) 2010-2011 Canonical Ltd. All Rights Reserved
+#  Copyright (C) 2010-2012 Canonical Ltd. All Rights Reserved
 #
 #  This file is part of txzookeeper.
 #