diff --git a/pgraph/actions.go b/pgraph/actions.go
index 9e811f40..dcf231f6 100644
--- a/pgraph/actions.go
+++ b/pgraph/actions.go
@@ -289,6 +289,9 @@ func (g *Graph) Worker(v *Vertex) error {
 	// the Watch() function about which graph it is
 	// running on, which isolates things nicely...
 	obj := v.Res
+	// TODO: is there a better system for the `Watching` flag?
+	obj.SetWatching(true)
+	defer obj.SetWatching(false)
 	processChan := make(chan event.Event)
 	go func() {
 		running := false
diff --git a/resources/exec.go b/resources/exec.go
index 11a25b68..b6743b7d 100644
--- a/resources/exec.go
+++ b/resources/exec.go
@@ -110,11 +110,6 @@ func (obj *ExecRes) BufioChanScanner(scanner *bufio.Scanner) (chan string, chan
 
 // Watch is the primary listener for this resource and it outputs events.
 func (obj *ExecRes) Watch(processChan chan event.Event) error {
-	if obj.IsWatching() {
-		return nil
-	}
-	obj.SetWatching(true)
-	defer obj.SetWatching(false)
 	cuid := obj.Converger() // get the converger uid used to report status
 
 	var send = false // send event?
diff --git a/resources/file.go b/resources/file.go
index 1c03b5da..8f2737aa 100644
--- a/resources/file.go
+++ b/resources/file.go
@@ -141,11 +141,6 @@ func (obj *FileRes) Validate() error {
 // must be restarted. On a clean exit it returns nil.
 // FIXME: Also watch the source directory when using obj.Source !!!
 func (obj *FileRes) Watch(processChan chan event.Event) error {
-	if obj.IsWatching() {
-		return nil // TODO: should this be an error?
-	}
-	obj.SetWatching(true)
-	defer obj.SetWatching(false)
 	cuid := obj.Converger() // get the converger uid used to report status
 
 	var err error
diff --git a/resources/hostname.go b/resources/hostname.go
index d235a130..552166eb 100644
--- a/resources/hostname.go
+++ b/resources/hostname.go
@@ -108,11 +108,6 @@ func (obj *HostnameRes) Validate() error {
 
 // Watch is the primary listener for this resource and it outputs events.
 func (obj *HostnameRes) Watch(processChan chan event.Event) error {
-	if obj.IsWatching() {
-		return nil // TODO: should this be an error?
-	}
-	obj.SetWatching(true)
-	defer obj.SetWatching(false)
 	cuid := obj.Converger() // get the converger uid used to report status
 
 	// if we share the bus with others, we will get each others messages!!
diff --git a/resources/msg.go b/resources/msg.go
index 64e29a9d..54b703db 100644
--- a/resources/msg.go
+++ b/resources/msg.go
@@ -135,11 +135,6 @@ func (obj *MsgRes) journalPriority() journal.Priority {
 
 // Watch is the primary listener for this resource and it outputs events.
 func (obj *MsgRes) Watch(processChan chan event.Event) error {
-	if obj.IsWatching() {
-		return nil
-	}
-	obj.SetWatching(true)
-	defer obj.SetWatching(false)
 	cuid := obj.Converger() // get the converger uid used to report status
 
 	// notify engine that we're running
diff --git a/resources/noop.go b/resources/noop.go
index 3e98c098..c77e640d 100644
--- a/resources/noop.go
+++ b/resources/noop.go
@@ -59,11 +59,6 @@ func (obj *NoopRes) Validate() error {
 
 // Watch is the primary listener for this resource and it outputs events.
 func (obj *NoopRes) Watch(processChan chan event.Event) error {
-	if obj.IsWatching() {
-		return nil // TODO: should this be an error?
-	}
-	obj.SetWatching(true)
-	defer obj.SetWatching(false)
 	cuid := obj.Converger() // get the converger uid used to report status
 
 	// notify engine that we're running
diff --git a/resources/nspawn.go b/resources/nspawn.go
index 58a0a884..5a6fa043 100644
--- a/resources/nspawn.go
+++ b/resources/nspawn.go
@@ -95,11 +95,6 @@ func (obj *NspawnRes) Validate() error {
 
 // Watch for state changes and sends a message to the bus if there is a change
 func (obj *NspawnRes) Watch(processChan chan event.Event) error {
-	if obj.IsWatching() {
-		return nil
-	}
-	obj.SetWatching(true)
-	defer obj.SetWatching(false)
 	cuid := obj.Converger() // get the converger uid used to report status
 
 	// this resource depends on systemd ensure that it's running
diff --git a/resources/password.go b/resources/password.go
index 26bd0dad..06ffba5e 100644
--- a/resources/password.go
+++ b/resources/password.go
@@ -168,11 +168,6 @@ Loop:
 
 // Watch is the primary listener for this resource and it outputs events.
 func (obj *PasswordRes) Watch(processChan chan event.Event) error {
-	if obj.IsWatching() {
-		return nil // TODO: should this be an error?
-	}
-	obj.SetWatching(true)
-	defer obj.SetWatching(false)
 	cuid := obj.Converger() // get the converger uid used to report status
 
 	var err error
diff --git a/resources/pkg.go b/resources/pkg.go
index 7b176852..497067eb 100644
--- a/resources/pkg.go
+++ b/resources/pkg.go
@@ -109,11 +109,6 @@ func (obj *PkgRes) Validate() error {
 // TODO: https://github.com/hughsie/PackageKit/issues/109
 // TODO: https://github.com/hughsie/PackageKit/issues/110
 func (obj *PkgRes) Watch(processChan chan event.Event) error {
-	if obj.IsWatching() {
-		return nil
-	}
-	obj.SetWatching(true)
-	defer obj.SetWatching(false)
 	cuid := obj.Converger() // get the converger uid used to report status
 
 	bus := packagekit.NewBus()
diff --git a/resources/resources.go b/resources/resources.go
index 1eefd351..959f6bb6 100644
--- a/resources/resources.go
+++ b/resources/resources.go
@@ -280,12 +280,12 @@ func (obj *BaseRes) AssociateData(data *Data) {
 	obj.debug = data.Debug
 }
 
-// IsWatching tells us if the Watch() function is running.
+// IsWatching tells us if the Worker() function is running.
 func (obj *BaseRes) IsWatching() bool {
 	return obj.watching
 }
 
-// SetWatching stores the status of if the Watch() function is running.
+// SetWatching stores the status of if the Worker() function is running.
 func (obj *BaseRes) SetWatching(b bool) {
 	obj.watching = b
 }
diff --git a/resources/svc.go b/resources/svc.go
index cb2b7c36..f58c0b9f 100644
--- a/resources/svc.go
+++ b/resources/svc.go
@@ -75,11 +75,6 @@ func (obj *SvcRes) Validate() error {
 
 // Watch is the primary listener for this resource and it outputs events.
 func (obj *SvcRes) Watch(processChan chan event.Event) error {
-	if obj.IsWatching() {
-		return nil
-	}
-	obj.SetWatching(true)
-	defer obj.SetWatching(false)
 	cuid := obj.Converger() // get the converger uid used to report status
 
 	// obj.Name: svc name
diff --git a/resources/timer.go b/resources/timer.go
index 0f022298..2fc9ca6c 100644
--- a/resources/timer.go
+++ b/resources/timer.go
@@ -74,11 +74,6 @@ func (obj *TimerRes) newTicker() *time.Ticker {
 
 // Watch is the primary listener for this resource and it outputs events.
 func (obj *TimerRes) Watch(processChan chan event.Event) error {
-	if obj.IsWatching() {
-		return nil
-	}
-	obj.SetWatching(true)
-	defer obj.SetWatching(false)
 	cuid := obj.Converger() // get the converger uid used to report status
 
 	// create a time.Ticker for the given interval
diff --git a/resources/virt.go b/resources/virt.go
index c67ac0ea..5e7c6e34 100644
--- a/resources/virt.go
+++ b/resources/virt.go
@@ -132,11 +132,6 @@ func (obj *VirtRes) connect() (conn libvirt.VirConnection, err error) {
 
 // Watch is the primary listener for this resource and it outputs events.
 func (obj *VirtRes) Watch(processChan chan event.Event) error {
-	if obj.IsWatching() {
-		return nil
-	}
-	obj.SetWatching(true)
-	defer obj.SetWatching(false)
 	cuid := obj.Converger() // get the converger uid used to report status
 
 	conn, err := obj.connect()