Merge lp:~niemeyer/juju-core/presence-polishing into lp:~juju/juju-core/trunk

[Gustavo Niemeyer]

mstate/presence: bring it in line with mstate/watcher

This mirrors into mstate/presence the improvements made into
the mstate/watcher package, including several points made in
reviews.

It also addresses a bug in Alive (it could return false
improperly without errors).

https://codereview.appspot.com/6501114/

[Gustavo Niemeyer (niemeyer)]

Reviewers: mp+123627_code.launchpad.net,

Message:
Please take a look.

Description:
mstate/presence: bring it in line with mstate/watcher

This mirrors into mstate/presence the improvements made into
the mstate/watcher package, including several points made in
reviews.

It also addresses a bug in Alive (it could return false
improperly without errors).

https://code.launchpad.net/~niemeyer/juju-core/presence-polishing/+merge/123627

Requires:
https://code.launchpad.net/~niemeyer/juju-core/mstate-machine-watcher/+merge/123614

(do not edit description out of merge proposal)

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

Affected files:
   A [revision details]
   M mstate/machine.go
   M mstate/machine_test.go
   M mstate/open.go
   M mstate/presence/presence.go
   M mstate/presence/presence_test.go
   M mstate/state.go
   M mstate/unit.go
   M mstate/unit_test.go
   M mstate/watcher/watcher_test.go

[William Reade (fwereade)]

LGTM, only very vague suggestions :).

https://codereview.appspot.com/6501114/diff/1/mstate/machine.go
File mstate/machine.go (left):

https://codereview.appspot.com/6501114/diff/1/mstate/machine.go#oldcode88
mstate/machine.go:88: func (m *Machine) WaitAgentAlive(timeout
time.Duration) error {
Kinda irrelevant, but I still don't know what use case these
WaitAgentAlive methods fulfil.

https://codereview.appspot.com/6501114/diff/1/mstate/machine.go
File mstate/machine.go (right):

https://codereview.appspot.com/6501114/diff/1/mstate/machine.go#newcode114
mstate/machine.go:114: return fmt.Errorf("waiting for agent of machine
%v: watcher is dying", m)
ErrorContextf?

https://codereview.appspot.com/6501114/diff/1/mstate/presence/presence.go
File mstate/presence/presence.go (right):

https://codereview.appspot.com/6501114/diff/1/mstate/presence/presence.go#newcode240
mstate/presence/presence.go:240: // w.pending may get new requests as we
handle other requests.
I'm not sure I quite follow how this works. It seems that it's
specifically *only* while handling requests that pending can change --
and that therefore there's no opportunity for it to change in between a
failed loop test and the closing truncate -- but I had to think about it
for a little bit longer than I would prefer, and it has a little hint of
looking-wrong to it. Would it make sense to expand the comment a little?

https://codereview.appspot.com/6501114/diff/1/mstate/unit.go
File mstate/unit.go (right):

https://codereview.appspot.com/6501114/diff/1/mstate/unit.go#newcode257
mstate/unit.go:257: return fmt.Errorf("waiting for agent of unit %q:
watcher is dying", u)
Similar comments to Machine.WaitAgentAlive

https://codereview.appspot.com/6501114/

[Gustavo Niemeyer (niemeyer)]

Please take a look.

https://codereview.appspot.com/6501114/diff/1/mstate/machine.go
File mstate/machine.go (left):

https://codereview.appspot.com/6501114/diff/1/mstate/machine.go#oldcode88
mstate/machine.go:88: func (m *Machine) WaitAgentAlive(timeout
time.Duration) error {
On 2012/09/11 12:09:31, fwereade wrote:
> Kinda irrelevant, but I still don't know what use case these
WaitAgentAlive
> methods fulfil.

LOL.. that crossed my mind. It feels like something clever we thought of
but never used.

https://codereview.appspot.com/6501114/diff/1/mstate/machine.go
File mstate/machine.go (right):

https://codereview.appspot.com/6501114/diff/1/mstate/machine.go#newcode114
mstate/machine.go:114: return fmt.Errorf("waiting for agent of machine
%v: watcher is dying", m)
On 2012/09/11 12:09:31, fwereade wrote:
> ErrorContextf?

Good catch, done.

I've also refactored those methods a bit to reduce them.

https://codereview.appspot.com/6501114/diff/1/mstate/presence/presence.go
File mstate/presence/presence.go (right):

https://codereview.appspot.com/6501114/diff/1/mstate/presence/presence.go#newcode240
mstate/presence/presence.go:240: // w.pending may get new requests as we
handle other requests.
On 2012/09/11 12:09:31, fwereade wrote:
> I'm not sure I quite follow how this works. It seems that it's
specifically
> *only* while handling requests that pending can change --

Events dispatched from the syncing procedure go into pending too. I'm
happy to expand the comment, but do you have any suggestions that would
make it clarify the misunderstanding you have/had?

https://codereview.appspot.com/6501114/

[William Reade (fwereade)]

LGTM

https://codereview.appspot.com/6501114/diff/1/mstate/machine.go
File mstate/machine.go (right):

https://codereview.appspot.com/6501114/diff/1/mstate/machine.go#newcode114
mstate/machine.go:114: return fmt.Errorf("waiting for agent of machine
%v: watcher is dying", m)
On 2012/09/11 13:27:15, niemeyer wrote:
> On 2012/09/11 12:09:31, fwereade wrote:
> > ErrorContextf?

> Good catch, done.

> I've also refactored those methods a bit to reduce them.

Very neat :)

https://codereview.appspot.com/6501114/diff/1/mstate/presence/presence.go
File mstate/presence/presence.go (right):

https://codereview.appspot.com/6501114/diff/1/mstate/presence/presence.go#newcode240
mstate/presence/presence.go:240: // w.pending may get new requests as we
handle other requests.
On 2012/09/11 13:27:15, niemeyer wrote:
> On 2012/09/11 12:09:31, fwereade wrote:
> > I'm not sure I quite follow how this works. It seems that it's
specifically
> > *only* while handling requests that pending can change --

> Events dispatched from the syncing procedure go into pending too. I'm
happy to
> expand the comment, but do you have any suggestions that would make it
clarify
> the misunderstanding you have/had?

Everything I can think of is a more cumbersome rewording of what you
already have -- best just forget I said anything :)

https://codereview.appspot.com/6501114/

[Gustavo Niemeyer (niemeyer)]

*** Submitted:

mstate/presence: bring it in line with mstate/watcher

This mirrors into mstate/presence the improvements made into
the mstate/watcher package, including several points made in
reviews.

It also addresses a bug in Alive (it could return false
improperly without errors).

R=fwereade
CC=
https://codereview.appspot.com/6501114

https://codereview.appspot.com/6501114/

=== modified file 'mstate/machine.go'
--- mstate/machine.go	2012-09-10 15:29:01 +0000
+++ mstate/machine.go	2012-09-11 13:29:53 +0000
@@ -80,35 +80,29 @@
 }
 
 // AgentAlive returns whether the respective remote agent is alive.
-func (m *Machine) AgentAlive() bool {
-	return m.st.presencew.Alive(m.globalKey())
+func (m *Machine) AgentAlive() (bool, error) {
+	return m.st.pwatcher.Alive(m.globalKey())
 }
 
 // WaitAgentAlive blocks until the respective agent is alive.
-func (m *Machine) WaitAgentAlive(timeout time.Duration) error {
+func (m *Machine) WaitAgentAlive(timeout time.Duration) (err error) {
+	defer trivial.ErrorContextf(&err, "waiting for agent of machine %v", m)
 	ch := make(chan presence.Change)
-	m.st.presencew.Add(m.globalKey(), ch)
-	defer m.st.presencew.Remove(m.globalKey(), ch)
-	// Initial check.
-	select {
-	case change := <-ch:
-		if change.Alive {
-			return nil
-		}
-	case <-time.After(timeout):
-		return fmt.Errorf("waiting for agent of machine %v: still not alive after timeout", m)
-	}
-	// Hasn't been alive, so now wait for change.
-	select {
-	case change := <-ch:
-		if change.Alive {
-			return nil
-		}
-		panic(fmt.Sprintf("presence reported dead status twice in a row for machine %v", m))
-	case <-time.After(timeout):
-		return fmt.Errorf("waiting for agent of machine %v: still not alive after timeout", m)
-	}
-	panic("unreachable")
+	m.st.pwatcher.Watch(m.globalKey(), ch)
+	defer m.st.pwatcher.Unwatch(m.globalKey(), ch)
+	for i := 0; i < 2; i++ {
+		select {
+		case change := <-ch:
+			if change.Alive {
+				return nil
+			}
+		case <-time.After(timeout):
+			return fmt.Errorf("still not alive after timeout")
+		case <-m.st.pwatcher.Dying():
+			return m.st.pwatcher.Err()
+		}
+	}
+	panic(fmt.Sprintf("presence reported dead status twice in a row for machine %v", m))
 }
 
 // SetAgentAlive signals that the agent for machine m is alive. 

=== modified file 'mstate/machine_test.go'
--- mstate/machine_test.go	2012-09-10 15:29:01 +0000
+++ mstate/machine_test.go	2012-09-11 13:29:53 +0000
@@ -23,7 +23,8 @@
 }
 
 func (s *MachineSuite) TestMachineSetAgentAlive(c *C) {
-	alive := s.machine.AgentAlive()
+	alive, err := s.machine.AgentAlive()
+	c.Assert(err, IsNil)
 	c.Assert(alive, Equals, false)
 
 	pinger, err := s.machine.SetAgentAlive()
@@ -31,37 +32,39 @@
 	c.Assert(pinger, Not(IsNil))
 	defer pinger.Stop()
 
-	s.State.ForcePresenceRefresh()
-	alive = s.machine.AgentAlive()
+	s.State.Sync()
+	alive, err = s.machine.AgentAlive()
+	c.Assert(err, IsNil)
 	c.Assert(alive, Equals, true)
 }
 
 func (s *MachineSuite) TestMachineWaitAgentAlive(c *C) {
 	// test -gocheck.f TestMachineWaitAgentAlive
 	timeout := 5 * time.Second
-	alive := s.machine.AgentAlive()
+	alive, err := s.machine.AgentAlive()
+	c.Assert(err, IsNil)
 	c.Assert(alive, Equals, false)
 
-	s.State.ForcePresenceRefresh()
-	err := s.machine.WaitAgentAlive(timeout)
+	s.State.StartSync()
+	err = s.machine.WaitAgentAlive(timeout)
 	c.Assert(err, ErrorMatches, `waiting for agent of machine 0: still not alive after timeout`)
 
 	pinger, err := s.machine.SetAgentAlive()
 	c.Assert(err, IsNil)
 
-	s.State.ForcePresenceRefresh()
+	s.State.StartSync()
 	err = s.machine.WaitAgentAlive(timeout)
 	c.Assert(err, IsNil)
 
-	alive = s.machine.AgentAlive()
+	alive, err = s.machine.AgentAlive()
 	c.Assert(err, IsNil)
 	c.Assert(alive, Equals, true)
 
 	err = pinger.Kill()
 	c.Assert(err, IsNil)
 
-	s.State.ForcePresenceRefresh()
-	alive = s.machine.AgentAlive()
+	s.State.Sync()
+	alive, err = s.machine.AgentAlive()
 	c.Assert(err, IsNil)
 	c.Assert(alive, Equals, false)
 }

=== modified file 'mstate/open.go'
--- mstate/open.go	2012-09-11 08:42:28 +0000
+++ mstate/open.go	2012-09-11 13:29:53 +0000
@@ -50,7 +50,7 @@
 	st.runner = txn.NewRunner(db.C("txns"))
 	st.runner.ChangeLog(db.C("txns.log"))
 	st.watcher = watcher.New(db.C("txns.log"))
-	st.presencew = presence.NewWatcher(pdb.C("presence"))
+	st.pwatcher = presence.NewWatcher(pdb.C("presence"))
 	for _, index := range indexes {
 		err = st.relations.EnsureIndex(index)
 		if err != nil {
@@ -61,8 +61,8 @@
 }
 
 func (st *State) Close() error {
-	err1 := st.presencew.Stop()
-	err2 := st.watcher.Stop()
+	err1 := st.watcher.Stop()
+	err2 := st.pwatcher.Stop()
 	st.db.Session.Close()
 	for _, err := range []error{err1, err2} {
 		if err != nil {

=== modified file 'mstate/presence/presence.go'
--- mstate/presence/presence.go	2012-09-06 22:12:07 +0000
+++ mstate/presence/presence.go	2012-09-11 13:29:53 +0000
@@ -58,7 +58,7 @@
 	beingKey map[int64]string
 	beingSeq map[string]int64
 
-	// watches has the per-key observer channels from Add/Remove.
+	// watches has the per-key observer channels from Watch/Unwatch.
 	watches map[string][]chan<- Change
 
 	// pending contains all the events to be dispatched to the watcher
@@ -70,13 +70,12 @@
 	// the the gorotuine loop.
 	request chan interface{}
 
-	// refreshed contains pending ForceRefresh done channels
-	// that are waiting for the completion notice.
-	refreshed []chan bool
+	// syncDone contains pending done channels from sync requests.
+	syncDone []chan bool
 
-	// next will dispatch when it's time to refresh the database
+	// next will dispatch when it's time to sync the database
 	// knowledge. It's maintained here so that ForceRefresh
-	// can manipulate it to force a refresh sooner.
+	// can manipulate it to force a sync sooner.
 	next <-chan time.Time
 }
 
@@ -116,17 +115,31 @@
 	return w.tomb.Wait()
 }
 
-type reqAdd struct {
-	key string
-	ch  chan<- Change
-}
-
-type reqRemove struct {
-	key string
-	ch  chan<- Change
-}
-
-type reqRefresh struct {
+// Dying returns a channel that is closed when the watcher is stopping
+// due to an error or because Stop was called explicitly.
+func (w *Watcher) Dying() <-chan struct{} {
+	return w.tomb.Dying()
+}
+
+// Err returns the error with which the watcher stopped.
+// It returns nil if the watcher stopped cleanly, tomb.ErrStillAlive
+// if the watcher is still running properly, or the respective error
+// if the watcher is terminating or has terminated with an error.
+func (w *Watcher) Err() error {
+	return w.tomb.Err()
+}
+
+type reqWatch struct {
+	key string
+	ch  chan<- Change
+}
+
+type reqUnwatch struct {
+	key string
+	ch  chan<- Change
+}
+
+type reqSync struct {
 	done chan bool
 }
 
@@ -135,55 +148,55 @@
 	result chan bool
 }
 
-// Add includes key into w for liveness monitoring. An event will
+func (w *Watcher) sendReq(req interface{}) {
+	select {
+	case w.request <- req:
+	case <-w.tomb.Dying():
+	}
+}
+
+// Watch starts watching the liveness of key. An event will
 // be sent onto ch to report the initial status for the key, and
 // from then on a new event will be sent whenever a change is
 // detected. Change values sent to the channel must be consumed,
 // or the whole watcher will blocked.
-func (w *Watcher) Add(key string, ch chan<- Change) {
-	select {
-	case w.request <- reqAdd{key, ch}:
-	case <-w.tomb.Dying():
-	}
-}
-
-// Remove removes key and ch from liveness monitoring.
-func (w *Watcher) Remove(key string, ch chan<- Change) {
-	select {
-	case w.request <- reqRemove{key, ch}:
-	case <-w.tomb.Dying():
-	}
-}
-
-// ForceRefresh forces a synchronous refresh of the watcher knowledge.
-// It blocks until the database state has been loaded and the events
-// have been prepared, but unblocks before changes are sent onto the
-// registered channels.
-func (w *Watcher) ForceRefresh() {
+func (w *Watcher) Watch(key string, ch chan<- Change) {
+	w.sendReq(reqWatch{key, ch})
+}
+
+// Unwatch stops watching the liveness of key via ch.
+func (w *Watcher) Unwatch(key string, ch chan<- Change) {
+	w.sendReq(reqUnwatch{key, ch})
+}
+
+// StartSync forces the watcher to load new events from the database.
+func (w *Watcher) StartSync() {
+	w.sendReq(reqSync{nil})
+}
+
+// Sync forces the watcher to load new events from the database and blocks
+// until all events have been dispatched.
+func (w *Watcher) Sync() {
 	done := make(chan bool)
-	select {
-	case w.request <- reqRefresh{done}:
-	case <-w.tomb.Dying():
-	}
+	w.sendReq(reqSync{done})
 	select {
 	case <-done:
 	case <-w.tomb.Dying():
 	}
 }
 
-// Alive returns whether the key is currently considered alive by w.
-func (w *Watcher) Alive(key string) bool {
+// Alive returns whether the key is currently considered alive by w,
+// or an error in case the watcher is dying.
+func (w *Watcher) Alive(key string) (bool, error) {
 	result := make(chan bool, 1)
-	select {
-	case w.request <- reqAlive{key, result}:
-	case <-w.tomb.Dying():
-	}
+	w.sendReq(reqAlive{key, result})
 	var alive bool
 	select {
 	case alive = <-result:
 	case <-w.tomb.Dying():
+		return false, fmt.Errorf("cannot check liveness: watcher is dying")
 	}
-	return alive
+	return alive, nil
 }
 
 // period is the length of each time slot in seconds.
@@ -205,18 +218,19 @@
 			return tomb.ErrDying
 		case <-w.next:
 			w.next = time.After(time.Duration(period) * time.Second)
-			refreshed := w.refreshed
-			w.refreshed = nil
-			if err := w.refresh(); err != nil {
+			syncDone := w.syncDone
+			w.syncDone = nil
+			if err := w.sync(); err != nil {
 				return err
 			}
-			for _, done := range refreshed {
+			w.flush()
+			for _, done := range syncDone {
 				close(done)
 			}
 		case req := <-w.request:
 			w.handle(req)
+			w.flush()
 		}
-		w.flush()
 	}
 	return nil
 }
@@ -224,20 +238,18 @@
 // flush sends all pending events to their respective channels.
 func (w *Watcher) flush() {
 	// w.pending may get new requests as we handle other requests.
-	i := 0
-	for i < len(w.pending) {
+	for i := 0; i < len(w.pending); i++ {
 		e := &w.pending[i]
-		if e.ch == nil {
-			i++ // Removed meanwhile.
-			continue
-		}
-		select {
-		case <-w.tomb.Dying():
-			return
-		case req := <-w.request:
-			w.handle(req)
-		case e.ch <- Change{e.key, e.alive}:
-			i++
+		for e.ch != nil {
+			select {
+			case <-w.tomb.Dying():
+				return
+			case req := <-w.request:
+				w.handle(req)
+				continue
+			case e.ch <- Change{e.key, e.alive}:
+			}
+			break
 		}
 	}
 	w.pending = w.pending[:0]
@@ -248,10 +260,12 @@
 func (w *Watcher) handle(req interface{}) {
 	log.Debugf("presence: got request: %#v", req)
 	switch r := req.(type) {
-	case reqRefresh:
+	case reqSync:
 		w.next = time.After(0)
-		w.refreshed = append(w.refreshed, r.done)
-	case reqAdd:
+		if r.done != nil {
+			w.syncDone = append(w.syncDone, r.done)
+		}
+	case reqWatch:
 		for _, ch := range w.watches[r.key] {
 			if ch == r.ch {
 				panic("adding channel twice for same key")
@@ -260,7 +274,7 @@
 		w.watches[r.key] = append(w.watches[r.key], r.ch)
 		_, alive := w.beingSeq[r.key]
 		w.pending = append(w.pending, event{r.ch, r.key, alive})
-	case reqRemove:
+	case reqUnwatch:
 		watches := w.watches[r.key]
 		for i, ch := range watches {
 			if ch == r.ch {
@@ -294,11 +308,11 @@
 	Dead  map[string]int64 ",omitempty"
 }
 
-// refresh updates the watcher knowledge from the database, and
+// sync updates the watcher knowledge from the database, and
 // queues events to observing channels. It fetches the last two time
 // slots and compares the union of both to the in-memory state.
-func (w *Watcher) refresh() error {
-	log.Debugf("presence: refreshing watcher knowledge from database...")
+func (w *Watcher) sync() error {
+	log.Debugf("presence: synchronizing watcher knowledge with database...")
 	slot := timeSlot(time.Now(), w.delta)
 	var ping []pingInfo
 	err := w.pings.Find(bson.D{{"$or", []pingInfo{{Slot: slot}, {Slot: slot - period}}}}).All(&ping)

=== modified file 'mstate/presence/presence_test.go'
--- mstate/presence/presence_test.go	2012-09-06 22:12:07 +0000
+++ mstate/presence/presence_test.go	2012-09-11 13:29:53 +0000
@@ -5,6 +5,7 @@
 	. "launchpad.net/gocheck"
 	"launchpad.net/juju-core/mstate/presence"
 	"launchpad.net/juju-core/testing"
+	"launchpad.net/tomb"
 	"strconv"
 	stdtesting "testing"
 	"time"
@@ -71,6 +72,40 @@
 	}
 }
 
+func assertAlive(c *C, w *presence.Watcher, key string, alive bool) {
+	alive, err := w.Alive("a")
+	c.Assert(err, IsNil)
+	c.Assert(alive, Equals, alive)
+}
+
+func (s *PresenceSuite) TestErrAndDying(c *C) {
+	w := presence.NewWatcher(s.presence)
+	defer w.Stop()
+
+	c.Assert(w.Err(), Equals, tomb.ErrStillAlive)
+	select {
+	case <-w.Dying():
+		c.Fatalf("Dying channel fired unexpectedly")
+	default:
+	}
+	c.Assert(w.Stop(), IsNil)
+	c.Assert(w.Err(), IsNil)
+	select {
+	case <-w.Dying():
+	default:
+		c.Fatalf("Dying channel should have fired")
+	}
+}
+
+func (s *PresenceSuite) TestAliveError(c *C) {
+	w := presence.NewWatcher(s.presence)
+	c.Assert(w.Stop(), IsNil)
+
+	alive, err := w.Alive("a")
+	c.Assert(err, ErrorMatches, ".*: watcher is dying")
+	c.Assert(alive, Equals, false)
+}
+
 func (s *PresenceSuite) TestWorkflow(c *C) {
 	w := presence.NewWatcher(s.presence)
 	pa := presence.NewPinger(s.presence, "a")
@@ -79,61 +114,61 @@
 	defer pa.Stop()
 	defer pb.Stop()
 
-	c.Assert(w.Alive("a"), Equals, false)
-	c.Assert(w.Alive("b"), Equals, false)
+	assertAlive(c, w, "a", false)
+	assertAlive(c, w, "b", false)
 
 	// Buffer one entry to avoid blocking the watcher here.
 	cha := make(chan presence.Change, 1)
 	chb := make(chan presence.Change, 1)
-	w.Add("a", cha)
-	w.Add("b", chb)
+	w.Watch("a", cha)
+	w.Watch("b", chb)
 
 	// Initial events with current status.
 	assertChange(c, cha, presence.Change{"a", false})
 	assertChange(c, chb, presence.Change{"b", false})
 
-	w.ForceRefresh()
+	w.StartSync()
 	assertNoChange(c, cha)
 	assertNoChange(c, chb)
 
 	c.Assert(pa.Start(), IsNil)
 
-	w.ForceRefresh()
+	w.StartSync()
 	assertChange(c, cha, presence.Change{"a", true})
 	assertNoChange(c, cha)
 	assertNoChange(c, chb)
 
-	c.Assert(w.Alive("a"), Equals, true)
-	c.Assert(w.Alive("b"), Equals, false)
+	assertAlive(c, w, "a", true)
+	assertAlive(c, w, "b", false)
 
 	// Changes while the channel is out are not observed.
-	w.Remove("a", cha)
+	w.Unwatch("a", cha)
 	assertNoChange(c, cha)
 	pa.Kill()
-	w.ForceRefresh()
+	w.Sync()
 	pa = presence.NewPinger(s.presence, "a")
 	pa.Start()
-	w.ForceRefresh()
+	w.StartSync()
 	assertNoChange(c, cha)
 
 	// We can still query it manually, though.
-	c.Assert(w.Alive("a"), Equals, true)
-	c.Assert(w.Alive("b"), Equals, false)
+	assertAlive(c, w, "a", true)
+	assertAlive(c, w, "b", false)
 
 	// Initial positive event. No refresh needed.
-	w.Add("a", cha)
+	w.Watch("a", cha)
 	assertChange(c, cha, presence.Change{"a", true})
 
 	c.Assert(pb.Start(), IsNil)
 
-	w.ForceRefresh()
+	w.StartSync()
 	assertChange(c, chb, presence.Change{"b", true})
 	assertNoChange(c, cha)
 	assertNoChange(c, chb)
 
 	c.Assert(pa.Stop(), IsNil)
 
-	w.ForceRefresh()
+	w.StartSync()
 	assertNoChange(c, cha)
 	assertNoChange(c, chb)
 
@@ -141,7 +176,7 @@
 	c.Assert(pa.Kill(), IsNil)
 	c.Assert(pb.Kill(), IsNil)
 
-	w.ForceRefresh()
+	w.StartSync()
 	assertChange(c, cha, presence.Change{"a", false})
 	assertChange(c, chb, presence.Change{"b", false})
 
@@ -172,11 +207,11 @@
 	c.Logf("Checking who's still alive...")
 	w := presence.NewWatcher(s.presence)
 	defer w.Stop()
-	w.ForceRefresh()
+	w.Sync()
 	ch := make(chan presence.Change)
 	for i := 0; i < N; i++ {
 		k := strconv.Itoa(i)
-		w.Add(k, ch)
+		w.Watch(k, ch)
 		if i%2 == 0 {
 			assertChange(c, ch, presence.Change{k, true})
 		} else {
@@ -192,26 +227,26 @@
 	defer p.Stop()
 
 	ch := make(chan presence.Change)
-	w.Add("a", ch)
+	w.Watch("a", ch)
 	assertChange(c, ch, presence.Change{"a", false})
 
 	c.Assert(p.Start(), IsNil)
-	w.ForceRefresh()
+	w.StartSync()
 	assertChange(c, ch, presence.Change{"a", true})
 
 	// Still alive in previous slot.
 	presence.FakeTimeSlot(1)
-	w.ForceRefresh()
+	w.StartSync()
 	assertNoChange(c, ch)
 
 	// Two last slots are empty.
 	presence.FakeTimeSlot(2)
-	w.ForceRefresh()
+	w.StartSync()
 	assertChange(c, ch, presence.Change{"a", false})
 
 	// Already dead so killing isn't noticed.
 	p.Kill()
-	w.ForceRefresh()
+	w.StartSync()
 	assertNoChange(c, ch)
 }
 
@@ -225,7 +260,7 @@
 	defer p.Stop()
 
 	ch := make(chan presence.Change)
-	w.Add("a", ch)
+	w.Watch("a", ch)
 	assertChange(c, ch, presence.Change{"a", false})
 
 	// A single ping.
@@ -237,18 +272,18 @@
 	assertChange(c, ch, presence.Change{"a", true})
 }
 
-func (s *PresenceSuite) TestAddRemoveOnQueue(c *C) {
+func (s *PresenceSuite) TestWatchUnwatchOnQueue(c *C) {
 	w := presence.NewWatcher(s.presence)
 	ch := make(chan presence.Change)
 	for i := 0; i < 100; i++ {
 		key := strconv.Itoa(i)
 		c.Logf("Adding %q", key)
-		w.Add(key, ch)
+		w.Watch(key, ch)
 	}
 	for i := 1; i < 100; i += 2 {
 		key := strconv.Itoa(i)
 		c.Logf("Removing %q", key)
-		w.Remove(key, ch)
+		w.Unwatch(key, ch)
 	}
 	alive := make(map[string]bool)
 	for i := 0; i < 50; i++ {
@@ -288,10 +323,10 @@
 		stop := false
 		for !stop {
 			w := presence.NewWatcher(s.presence)
-			w.ForceRefresh()
-			alive := w.Alive("a")
+			w.Sync()
+			alive, err := w.Alive("a")
 			c.Check(w.Stop(), IsNil)
-			if !c.Check(alive, Equals, true) {
+			if !c.Check(err, IsNil) || !c.Check(alive, Equals, true) {
 				break
 			}
 			select {
@@ -325,22 +360,87 @@
 	// Start p1 and let it go on.
 	c.Assert(p1.Start(), IsNil)
 
-	w.ForceRefresh()
-	c.Assert(w.Alive("a"), Equals, true)
+	w.Sync()
+	assertAlive(c, w, "a", true)
 
 	// Start and kill p2, which will temporarily
 	// invalidate p1 and set the key as dead.
 	c.Assert(p2.Start(), IsNil)
 	c.Assert(p2.Kill(), IsNil)
 
-	w.ForceRefresh()
-	c.Assert(w.Alive("a"), Equals, false)
+	w.Sync()
+	assertAlive(c, w, "a", false)
 
 	// Wait for two periods, and check again. Since
 	// p1 is still alive, p2's death will expire and
 	// the key will come back.
 	time.Sleep(period * 2 * time.Second)
 
-	w.ForceRefresh()
-	c.Assert(w.Alive("a"), Equals, true)
+	w.Sync()
+	assertAlive(c, w, "a", true)
+}
+
+func (s *PresenceSuite) TestStartSync(c *C) {
+	w := presence.NewWatcher(s.presence)
+	p := presence.NewPinger(s.presence, "a")
+	defer w.Stop()
+	defer p.Stop()
+
+	ch := make(chan presence.Change)
+	w.Watch("a", ch)
+	assertChange(c, ch, presence.Change{"a", false})
+
+	c.Assert(p.Start(), IsNil)
+
+	done := make(chan bool)
+	go func() {
+		w.StartSync()
+		w.StartSync()
+		w.StartSync()
+		done <- true
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(100 * time.Millisecond):
+		c.Fatalf("StartSync failed to return")
+	}
+
+	assertChange(c, ch, presence.Change{"a", true})
+}
+
+func (s *PresenceSuite) TestSync(c *C) {
+	w := presence.NewWatcher(s.presence)
+	p := presence.NewPinger(s.presence, "a")
+	defer w.Stop()
+	defer p.Stop()
+
+	ch := make(chan presence.Change)
+	w.Watch("a", ch)
+	assertChange(c, ch, presence.Change{"a", false})
+
+	// Nothing to do here.
+	w.Sync()
+
+	c.Assert(p.Start(), IsNil)
+
+	done := make(chan bool)
+	go func() {
+		w.Sync()
+		done <- true
+	}()
+
+	select {
+	case <-done:
+		c.Fatalf("Sync returned too early")
+	case <-time.After(200 * time.Millisecond):
+	}
+
+	assertChange(c, ch, presence.Change{"a", true})
+
+	select {
+	case <-done:
+	case <-time.After(100 * time.Millisecond):
+		c.Fatalf("Sync failed to returned")
+	}
 }

=== modified file 'mstate/state.go'
--- mstate/state.go	2012-09-10 15:29:01 +0000
+++ mstate/state.go	2012-09-11 13:29:53 +0000
@@ -28,17 +28,17 @@
 // State represents the state of an environment
 // managed by juju.
 type State struct {
-	db         *mgo.Database
-	charms     *mgo.Collection
-	machines   *mgo.Collection
-	relations  *mgo.Collection
-	services   *mgo.Collection
-	settings   *mgo.Collection
-	units      *mgo.Collection
-	presence   *mgo.Collection
-	runner     *txn.Runner
-	watcher    *watcher.Watcher
-	presencew  *presence.Watcher
+	db        *mgo.Database
+	charms    *mgo.Collection
+	machines  *mgo.Collection
+	relations *mgo.Collection
+	services  *mgo.Collection
+	settings  *mgo.Collection
+	units     *mgo.Collection
+	presence  *mgo.Collection
+	runner    *txn.Runner
+	watcher   *watcher.Watcher
+	pwatcher  *presence.Watcher
 }
 
 func deadOnAbort(err error) error {
@@ -365,16 +365,16 @@
 	return newUnit(s, &doc), nil
 }
 
-// ForcePresenceRefresh forces a synchronous refresh of 
-// the presence watcher knowledge
-func (s *State) ForcePresenceRefresh() {
-	s.presencew.ForceRefresh()
-}
-
 // StartSync forces watchers to resynchronize their state with the
 // database immediately. This will happen periodically automatically.
 func (s *State) StartSync() {
-	// TODO Make presence more like watcher, add it here, and
-	// remove ForcePresenceRefresh.
 	s.watcher.StartSync()
+	s.pwatcher.StartSync()
+}
+
+// Sync forces watchers to resynchronize their state with the
+// database immediately, and waits until all events are known.
+func (s *State) Sync() {
+	s.watcher.Sync()
+	s.pwatcher.Sync()
 }

=== modified file 'mstate/unit.go'
--- mstate/unit.go	2012-09-06 16:54:22 +0000
+++ mstate/unit.go	2012-09-11 13:29:53 +0000
@@ -196,7 +196,10 @@
 	case UnitStopped:
 		return UnitStopped, "", nil
 	}
-	alive := u.AgentAlive()
+	alive, err := u.AgentAlive()
+	if err != nil {
+		return "", "", err
+	}
 	if !alive {
 		s = UnitDown
 	}
@@ -222,35 +225,29 @@
 }
 
 // AgentAlive returns whether the respective remote agent is alive.
-func (u *Unit) AgentAlive() bool {
-	return u.st.presencew.Alive(u.globalKey())
+func (u *Unit) AgentAlive() (bool, error) {
+	return u.st.pwatcher.Alive(u.globalKey())
 }
 
 // WaitAgentAlive blocks until the respective agent is alive.
-func (u *Unit) WaitAgentAlive(timeout time.Duration) error {
+func (u *Unit) WaitAgentAlive(timeout time.Duration) (err error) {
+	defer trivial.ErrorContextf(&err, "waiting for agent of unit %q", u)
 	ch := make(chan presence.Change)
-	u.st.presencew.Add(u.globalKey(), ch)
-	defer u.st.presencew.Remove(u.globalKey(), ch)
-	// Initial check.
-	select {
-	case change := <-ch:
-		if change.Alive {
-			return nil
-		}
-	case <-time.After(timeout):
-		return fmt.Errorf("waiting for agent of unit %q: still not alive after timeout", u)
-	}
-	// Hasn't been alive, so now wait for change.
-	select {
-	case change := <-ch:
-		if change.Alive {
-			return nil
-		}
-		panic(fmt.Sprintf("presence reported dead status twice in a row for unit %q", u))
-	case <-time.After(timeout):
-		return fmt.Errorf("waiting for agent of unit %q: still not alive after timeout", u)
-	}
-	panic("unreachable")
+	u.st.pwatcher.Watch(u.globalKey(), ch)
+	defer u.st.pwatcher.Unwatch(u.globalKey(), ch)
+	for i := 0; i < 2; i++ {
+		select {
+		case change := <-ch:
+			if change.Alive {
+				return nil
+			}
+		case <-time.After(timeout):
+			return fmt.Errorf("still not alive after timeout")
+		case <-u.st.pwatcher.Dying():
+			return u.st.pwatcher.Err()
+		}
+	}
+	panic(fmt.Sprintf("presence reported dead status twice in a row for unit %q", u))
 }
 
 // SetAgentAlive signals that the agent for unit u is alive. 

=== modified file 'mstate/unit_test.go'
--- mstate/unit_test.go	2012-09-06 17:04:54 +0000
+++ mstate/unit_test.go	2012-09-11 13:29:53 +0000
@@ -90,7 +90,7 @@
 		c.Assert(p.Kill(), IsNil)
 	}()
 
-	s.State.ForcePresenceRefresh()
+	s.State.StartSync()
 	status, info, err = s.unit.Status()
 	c.Assert(err, IsNil)
 	c.Assert(status, Equals, state.UnitStarted)
@@ -105,7 +105,8 @@
 }
 
 func (s *UnitSuite) TestUnitSetAgentAlive(c *C) {
-	alive := s.unit.AgentAlive()
+	alive, err := s.unit.AgentAlive()
+	c.Assert(err, IsNil)
 	c.Assert(alive, Equals, false)
 
 	pinger, err := s.unit.SetAgentAlive()
@@ -113,34 +114,38 @@
 	c.Assert(pinger, Not(IsNil))
 	defer pinger.Stop()
 
-	s.State.ForcePresenceRefresh()
-	alive = s.unit.AgentAlive()
+	s.State.Sync()
+	alive, err = s.unit.AgentAlive()
+	c.Assert(err, IsNil)
 	c.Assert(alive, Equals, true)
 }
 
 func (s *UnitSuite) TestUnitWaitAgentAlive(c *C) {
 	timeout := 5 * time.Second
-	alive := s.unit.AgentAlive()
+	alive, err := s.unit.AgentAlive()
+	c.Assert(err, IsNil)
 	c.Assert(alive, Equals, false)
 
-	err := s.unit.WaitAgentAlive(timeout)
+	err = s.unit.WaitAgentAlive(timeout)
 	c.Assert(err, ErrorMatches, `waiting for agent of unit "wordpress/0": still not alive after timeout`)
 
 	pinger, err := s.unit.SetAgentAlive()
 	c.Assert(err, IsNil)
 
-	s.State.ForcePresenceRefresh()
+	s.State.StartSync()
 	err = s.unit.WaitAgentAlive(timeout)
 	c.Assert(err, IsNil)
 
-	alive = s.unit.AgentAlive()
+	alive, err = s.unit.AgentAlive()
+	c.Assert(err, IsNil)
 	c.Assert(alive, Equals, true)
 
 	err = pinger.Kill()
 	c.Assert(err, IsNil)
 
-	s.State.ForcePresenceRefresh()
-	alive = s.unit.AgentAlive()
+	s.State.Sync()
+	alive, err = s.unit.AgentAlive()
+	c.Assert(err, IsNil)
 	c.Assert(alive, Equals, false)
 }
 

=== modified file 'mstate/watcher/watcher_test.go'
--- mstate/watcher/watcher_test.go	2012-09-10 16:19:57 +0000
+++ mstate/watcher/watcher_test.go	2012-09-11 13:29:53 +0000
@@ -440,9 +440,6 @@
 func (s *WatcherSuite) TestStartSync(c *C) {
 	s.w.Watch("test", "a", -1, s.ch)
 
-	// Nothing to do here.
-	s.w.StartSync()
-
 	revno := s.insert(c, "test", "a")
 
 	done := make(chan bool)
@@ -456,7 +453,7 @@
 	select {
 	case <-done:
 	case <-time.After(100 * time.Millisecond):
-		c.Fatalf("SyncStart failed to return")
+		c.Fatalf("StartSync failed to return")
 	}
 
 	assertChange(c, s.ch, watcher.Change{"test", "a", revno})
@@ -479,7 +476,7 @@
 	select {
 	case <-done:
 		c.Fatalf("Sync returned too early")
-	case <-time.After(500 * time.Millisecond):
+	case <-time.After(200 * time.Millisecond):
 	}
 
 	assertChange(c, s.ch, watcher.Change{"test", "a", revno})