process watches locally

currently watches use zab.flush. it would be nice to be able to handle watches locally.