Repository: godbus/dbus (go)

Issue:
Unable to correctly implement State Caching Proxy Pattern using dbusutil
Hi All,

*Background:*
I am working on an application that involves tracking the state of a DBus object. The object I am using provides a method to get its current state, as well as a signal to notify us whenever that state changes.

I would like to build a local object which tracks the state of the remote object and caches it for quick access. In the literature, I have seen this referred to as a a _State Caching Proxy pattern_. 

The guarantee I would like to achieve is that once the object has been initialised, the observable states of this local object will be a contiguous subsequence of the observable states of the remote object. That is, if the remote object transitions through states S(0), S(1) ... S(n) during the lifetime of the system, the local object will expose some contiguous subsequence of those states (for as the local object exists and is subscribed). 
Valid observation trace: S(3), S(4), S(5), S(6) ...
Invalid observation trace: S(4), S(3), S(5), S(6) ... (States received out of order.)
Invalid observation trace: S(3), S(5), S(6). (Missing state changes.)

If at the point the local caching proxy object is initialised, it exposes state S(3), I expect to be able to observe all future state changes (i.e. S(4), S(5), S(6), ..) in exactly that order without missing any changes *and* without any other state changes mixed in. 

The State Caching Proxy pattern is described in section 5.2 of Olli Salli's masters thesis:
"Building object-oriented software with the D-Bus messaging system"
https://www.utupub.fi/bitstream/handle/10024/78669/gradu2012Salli.pdf?sequence=1

As described in that thesis, the correct way of implementing this pattern is to subscribe for changes first and _then_ ask for the object's current state. Only once the current state is received, is the object considered initialised and are further state changes used.

*Problem*
Unfortunately, dbusutil presents two impediments to correctly implementing this pattern:
- Signals are not necessarily received in-order. The current implementation spins up a separate goroutine to dispatch each signal. There is no guarantee these goroutines will execute in-order (indeed, in our observation, for signals sent in close succession, the goroutines execute in arbitrary order. As such, we are receiving signals out of order.). 
- I have no way of identifying if a signal was received before or after the method reply that told us the initial state of the object. Both signals and method responses are handled in a somewhat asynchronous manner, using different channels and goroutines. Consequently, the original order of messages on the DBus is lost. This ordering is the only way in which we can ascertain if a state change signal is for an old state (prior to us querying the initial state) and should be discarded, or is for a new state.

To my knowledge, other D-Bus libraries such as GDBus, provide mechanisms to correctly implement this pattern by offering synchronous processing of D-Bus messages.

I would be open to contributing changes to fixing dbusutil to allow the above pattern, but it seems to me that it may require significant architectural changes (mostly for the second issue: exposing ordering between method responses and signals). I would therefore like to enquire what the maintainers of this package think about this issue.

Thanks in advance,
Patrick


Pull request description:
Allow race-free implementation of the State Caching Proxy Pattern.
This is a revised version of the code in Pull Request #190, which got muddied with other commits I made trying to diagnose issues in CI.

This change is intended to fix #187.

As before:
This is a prototype of my proposed solution, with tests. I am unsure about the naming of external interface members and whether additional changes should be made to the external interface (other than adding Signal.Sequence and Call.ResponseSequence).

If you are completely happy with this change, please merge, otherwise please provide feedback and I will address.

Thanks,
Patrick

Patch:
diff --git a/call.go b/call.go
index 2cb1890..b06b063 100644
--- a/call.go
+++ b/call.go
@@ -24,6 +24,15 @@ type Call struct {
 	// Holds the response once the call is done.
 	Body []interface{}
 
+	// ResponseSequence stores the sequence number of the DBus message containing
+	// the call response (or error). This can be compared to the sequence number
+	// of other call responses and signals on this connection to determine their
+	// relative ordering on the underlying DBus connection.
+	// For errors, ResponseSequence is populated only if the error came from a
+	// DBusMessage that was received or if there was an error receiving. In case of
+	// failure to make the call, ResponseSequence will be NoSequence.
+	ResponseSequence Sequence
+
 	// tracks context and canceler
 	ctx         context.Context
 	ctxCanceler context.CancelFunc
diff --git a/conn.go b/conn.go
index ad00ccb..3507aa6 100644
--- a/conn.go
+++ b/conn.go
@@ -363,6 +363,7 @@ func (conn *Conn) Hello() error {
 // inWorker runs in an own goroutine, reading incoming messages from the
 // transport and dispatching them appropiately.
 func (conn *Conn) inWorker() {
+	sequenceGen := newSequenceGenerator()
 	for {
 		msg, err := conn.ReadMessage()
 		if err != nil {
@@ -371,7 +372,7 @@ func (conn *Conn) inWorker() {
 				// anything but to shut down all stuff and returns errors to all
 				// pending replies.
 				conn.Close()
-				conn.calls.finalizeAllWithError(err)
+				conn.calls.finalizeAllWithError(sequenceGen, err)
 				return
 			}
 			// invalid messages are ignored
@@ -400,13 +401,14 @@ func (conn *Conn) inWorker() {
 		if conn.inInt != nil {
 			conn.inInt(msg)
 		}
+		sequence := sequenceGen.next()
 		switch msg.Type {
 		case TypeError:
-			conn.serialGen.RetireSerial(conn.calls.handleDBusError(msg))
+			conn.serialGen.RetireSerial(conn.calls.handleDBusError(sequence, msg))
 		case TypeMethodReply:
-			conn.serialGen.RetireSerial(conn.calls.handleReply(msg))
+			conn.serialGen.RetireSerial(conn.calls.handleReply(sequence, msg))
 		case TypeSignal:
-			conn.handleSignal(msg)
+			conn.handleSignal(sequence, msg)
 		case TypeMethodCall:
 			go conn.handleCall(msg)
 		}
@@ -414,7 +416,7 @@ func (conn *Conn) inWorker() {
 	}
 }
 
-func (conn *Conn) handleSignal(msg *Message) {
+func (conn *Conn) handleSignal(sequence Sequence, msg *Message) {
 	iface := msg.Headers[FieldInterface].value.(string)
 	member := msg.Headers[FieldMember].value.(string)
 	// as per http://dbus.freedesktop.org/doc/dbus-specification.html ,
@@ -440,10 +442,11 @@ func (conn *Conn) handleSignal(msg *Message) {
 		}
 	}
 	signal := &Signal{
-		Sender: sender,
-		Path:   msg.Headers[FieldPath].value.(ObjectPath),
-		Name:   iface + "." + member,
-		Body:   msg.Body,
+		Sender:   sender,
+		Path:     msg.Headers[FieldPath].value.(ObjectPath),
+		Name:     iface + "." + member,
+		Body:     msg.Body,
+		Sequence: sequence,
 	}
 	conn.signalHandler.DeliverSignal(iface, member, signal)
 }
@@ -658,10 +661,11 @@ func (e Error) Error() string {
 // Signal represents a D-Bus message of type Signal. The name member is given in
 // "interface.member" notation, e.g. org.freedesktop.D-Bus.NameLost.
 type Signal struct {
-	Sender string
-	Path   ObjectPath
-	Name   string
-	Body   []interface{}
+	Sender   string
+	Path     ObjectPath
+	Name     string
+	Body     []interface{}
+	Sequence Sequence
 }
 
 // transport is a D-Bus transport.
@@ -844,25 +848,25 @@ func (tracker *callTracker) track(sn uint32, call *Call) {
 	tracker.lck.Unlock()
 }
 
-func (tracker *callTracker) handleReply(msg *Message) uint32 {
+func (tracker *callTracker) handleReply(sequence Sequence, msg *Message) uint32 {
 	serial := msg.Headers[FieldReplySerial].value.(uint32)
 	tracker.lck.RLock()
 	_, ok := tracker.calls[serial]
 	tracker.lck.RUnlock()
 	if ok {
-		tracker.finalizeWithBody(serial, msg.Body)
+		tracker.finalizeWithBody(serial, sequence, msg.Body)
 	}
 	return serial
 }
 
-func (tracker *callTracker) handleDBusError(msg *Message) uint32 {
+func (tracker *callTracker) handleDBusError(sequence Sequence, msg *Message) uint32 {
 	serial := msg.Headers[FieldReplySerial].value.(uint32)
 	tracker.lck.RLock()
 	_, ok := tracker.calls[serial]
 	tracker.lck.RUnlock()
 	if ok {
 		name, _ := msg.Headers[FieldErrorName].value.(string)
-		tracker.finalizeWithError(serial, Error{name, msg.Body})
+		tracker.finalizeWithError(serial, sequence, Error{name, msg.Body})
 	}
 	return serial
 }
@@ -875,7 +879,7 @@ func (tracker *callTracker) handleSendError(msg *Message, err error) {
 	_, ok := tracker.calls[msg.serial]
 	tracker.lck.RUnlock()
 	if ok {
-		tracker.finalizeWithError(msg.serial, err)
+		tracker.finalizeWithError(msg.serial, NoSequence, err)
 	}
 }
 
@@ -890,7 +894,7 @@ func (tracker *callTracker) finalize(sn uint32) {
 	}
 }
 
-func (tracker *callTracker) finalizeWithBody(sn uint32, body []interface{}) {
+func (tracker *callTracker) finalizeWithBody(sn uint32, sequence Sequence, body []interface{}) {
 	tracker.lck.Lock()
 	c, ok := tracker.calls[sn]
 	if ok {
@@ -899,11 +903,12 @@ func (tracker *callTracker) finalizeWithBody(sn uint32, body []interface{}) {
 	tracker.lck.Unlock()
 	if ok {
 		c.Body = body
+		c.ResponseSequence = sequence
 		c.done()
 	}
 }
 
-func (tracker *callTracker) finalizeWithError(sn uint32, err error) {
+func (tracker *callTracker) finalizeWithError(sn uint32, sequence Sequence, err error) {
 	tracker.lck.Lock()
 	c, ok := tracker.calls[sn]
 	if ok {
@@ -912,11 +917,12 @@ func (tracker *callTracker) finalizeWithError(sn uint32, err error) {
 	tracker.lck.Unlock()
 	if ok {
 		c.Err = err
+		c.ResponseSequence = sequence
 		c.done()
 	}
 }
 
-func (tracker *callTracker) finalizeAllWithError(err error) {
+func (tracker *callTracker) finalizeAllWithError(sequenceGen *sequenceGenerator, err error) {
 	tracker.lck.Lock()
 	closedCalls := make([]*Call, 0, len(tracker.calls))
 	for sn := range tracker.calls {
@@ -926,6 +932,7 @@ func (tracker *callTracker) finalizeAllWithError(err error) {
 	tracker.lck.Unlock()
 	for _, call := range closedCalls {
 		call.Err = err
+		call.ResponseSequence = sequenceGen.next()
 		call.done()
 	}
 }
diff --git a/sequence.go b/sequence.go
new file mode 100644
index 0000000..89435d3
--- /dev/null
+++ b/sequence.go
@@ -0,0 +1,24 @@
+package dbus
+
+// Sequence represents the value of a monotonically increasing counter.
+type Sequence uint64
+
+const (
+	// NoSequence indicates the absence of a sequence value.
+	NoSequence Sequence = 0
+)
+
+// sequenceGenerator represents a monotonically increasing counter.
+type sequenceGenerator struct {
+	nextSequence Sequence
+}
+
+func (generator *sequenceGenerator) next() Sequence {
+	result := generator.nextSequence
+	generator.nextSequence++
+	return result
+}
+
+func newSequenceGenerator() *sequenceGenerator {
+	return &sequenceGenerator{nextSequence: 1}
+}
diff --git a/sequential_handler.go b/sequential_handler.go
new file mode 100644
index 0000000..3a8cd71
--- /dev/null
+++ b/sequential_handler.go
@@ -0,0 +1,120 @@
+package dbus
+
+import (
+	"container/list"
+	"sync"
+)
+
+// NewSequentialSignalHandler returns an instance of a new
+// signal handler that guarantees sequential processing of signals. It is a
+// guarantee of this signal handler that signals will be written to
+// channels in the order they are received on the DBus connection.
+func NewSequentialSignalHandler() SignalHandler {
+	return &sequentialSignalHandler{}
+}
+
+type sequentialSignalHandler struct {
+	mu      sync.RWMutex
+	closed  bool
+	signals []*sequentialSignalChannelData
+}
+
+func (sh *sequentialSignalHandler) DeliverSignal(intf, name string, signal *Signal) {
+	sh.mu.RLock()
+	defer sh.mu.RUnlock()
+	if sh.closed {
+		return
+	}
+	for _, scd := range sh.signals {
+		scd.deliver(signal)
+	}
+}
+
+func (sh *sequentialSignalHandler) Terminate() {
+	sh.mu.Lock()
+	defer sh.mu.Unlock()
+	if sh.closed {
+		return
+	}
+
+	for _, scd := range sh.signals {
+		scd.close()
+		close(scd.ch)
+	}
+	sh.closed = true
+	sh.signals = nil
+}
+
+func (sh *sequentialSignalHandler) AddSignal(ch chan<- *Signal) {
+	sh.mu.Lock()
+	defer sh.mu.Unlock()
+	if sh.closed {
+		return
+	}
+	sh.signals = append(sh.signals, &sequentialSignalChannelData{
+		queue: list.New(),
+		ch:    ch,
+		done:  make(chan struct{}),
+	})
+}
+
+func (sh *sequentialSignalHandler) RemoveSignal(ch chan<- *Signal) {
+	sh.mu.Lock()
+	defer sh.mu.Unlock()
+	if sh.closed {
+		return
+	}
+	for i := len(sh.signals) - 1; i >= 0; i-- {
+		if ch == sh.signals[i].ch {
+			sh.signals[i].close()
+			copy(sh.signals[i:], sh.signals[i+1:])
+			sh.signals[len(sh.signals)-1] = nil
+			sh.signals = sh.signals[:len(sh.signals)-1]
+		}
+	}
+}
+
+type sequentialSignalChannelData struct {
+	stateLock sync.Mutex
+	writeLock sync.Mutex
+	queue     *list.List
+	wg        sync.WaitGroup
+	ch        chan<- *Signal
+	done      chan struct{}
+}
+
+func (scd *sequentialSignalChannelData) deliver(signal *Signal) {
+	// Avoid blocking the main DBus message processing routine;
+	// queue signal to be dispatched later.
+	scd.stateLock.Lock()
+	scd.queue.PushBack(signal)
+	scd.stateLock.Unlock()
+
+	scd.wg.Add(1)
+	go scd.deferredDeliver()
+}
+
+func (scd *sequentialSignalChannelData) deferredDeliver() {
+	defer scd.wg.Done()
+
+	// Ensure only one goroutine is in this section at once, to
+	// make sure signals are sent over ch in the order they
+	// are in the queue.
+	scd.writeLock.Lock()
+	defer scd.writeLock.Unlock()
+
+	scd.stateLock.Lock()
+	elem := scd.queue.Front()
+	scd.queue.Remove(elem)
+	scd.stateLock.Unlock()
+
+	select {
+	case scd.ch <- elem.Value.(*Signal):
+	case <-scd.done:
+	}
+}
+
+func (scd *sequentialSignalChannelData) close() {
+	close(scd.done)
+	scd.wg.Wait() // wait until all spawned goroutines return
+}


Test patch:
diff --git a/conn_test.go b/conn_test.go
index 9360131..9ab34c7 100644
--- a/conn_test.go
+++ b/conn_test.go
@@ -3,9 +3,11 @@ package dbus
 import (
 	"context"
 	"encoding/binary"
+	"fmt"
 	"io"
 	"io/ioutil"
 	"log"
+	"sync"
 	"testing"
 	"time"
 )
@@ -197,7 +199,7 @@ func TestCloseChannelAfterRemoveSignal(t *testing.T) {
 			FieldPath:      MakeVariant(ObjectPath("/baz")),
 		},
 	}
-	bus.handleSignal(msg)
+	bus.handleSignal(Sequence(1), msg)
 
 	// Remove and close the signal channel
 	bus.RemoveSignal(ch)
@@ -256,6 +258,168 @@ func waitSignal(sigc <-chan *Signal, name string, timeout time.Duration) *Signal
 	}
 }
 
+const (
+	SCPPInterface         = "org.godbus.DBus.StatefulTest"
+	SCPPPath              = "/org/godbus/DBus/StatefulTest"
+	SCPPChangedSignalName = "Changed"
+	SCPPStateMethodName   = "State"
+)
+
+func TestStateCachingProxyPattern(t *testing.T) {
+	srv, err := ConnectSessionBus()
+	defer srv.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	conn, err := ConnectSessionBus(WithSignalHandler(NewSequentialSignalHandler()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+
+	serviceName := srv.Names()[0]
+	// message channel should have at least some buffering, to make sure Eavesdrop does not
+	// drop the message if nobody is currently trying to read from the channel.
+	messages := make(chan *Message, 1)
+	srv.Eavesdrop(messages)
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		if err := serverProcess(ctx, srv, messages, t); err != nil {
+			t.Errorf("error in server process: %v", err)
+			cancel()
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		if err := clientProcess(ctx, conn, serviceName, t); err != nil {
+			t.Errorf("error in client process: %v", err)
+		}
+		// Cancel the server process.
+		cancel()
+	}()
+	wg.Wait()
+}
+
+func clientProcess(ctx context.Context, conn *Conn, serviceName string, t *testing.T) error {
+	// Subscribe to state changes on the remote object
+	if err := conn.AddMatchSignal(
+		WithMatchInterface(SCPPInterface),
+		WithMatchMember(SCPPChangedSignalName),
+	); err != nil {
+		return err
+	}
+	channel := make(chan *Signal)
+	conn.Signal(channel)
+	t.Log("Subscribed to signals")
+
+	// Simulate unfavourable OS scheduling leading to a delay between subscription
+	// and querying for the current state.
+	time.Sleep(30 * time.Millisecond)
+
+	// Call .State() on the remote object to get its current state and store in observedStates[0].
+	obj := conn.Object(serviceName, SCPPPath)
+	observedStates := make([]uint64, 1)
+	call := obj.CallWithContext(ctx, SCPPInterface+"."+SCPPStateMethodName, 0)
+	if err := call.Store(&observedStates[0]); err != nil {
+		return err
+	}
+	t.Logf("Queried current state, got %v", observedStates[0])
+
+	// Populate observedStates[1...49] based on the state change signals,
+	// ignoring signals with a sequence number less than call.ResponseSequence so that we ignore past signals.
+	signalsProcessed := 0
+readSignals:
+	for {
+		select {
+		case signal := <-channel:
+			signalsProcessed++
+			if signal.Name == SCPPInterface+"."+SCPPChangedSignalName && signal.Sequence > call.ResponseSequence {
+				observedState := signal.Body[0].(uint64)
+				observedStates = append(observedStates, observedState)
+				// Observing at least 50 states gives us low probability that we received a contiguous subsequence of states 'by accident'
+				if len(observedStates) >= 50 {
+					break readSignals
+				}
+			}
+		case <-ctx.Done():
+			t.Logf("Context cancelled, client processed %v signals", signalsProcessed)
+			return ctx.Err()
+		}
+	}
+	t.Logf("client processed %v signals", signalsProcessed)
+
+	// Expect that we begun observing at least a few states in. This ensures the server was already emitting signals
+	// and makes it likely we simulated our race condition.
+	if observedStates[0] < 10 {
+		return fmt.Errorf("expected first state to be at least 10, got %v", observedStates[0])
+	}
+
+	t.Logf("Observed states: %v", observedStates)
+
+	// The observable states of the remote object were [1 ... (infinity)] during this test.
+	// This loop is intended to assert that our observed states are a contiguous subgrange [n ... n+49] for some n, i.e.
+	// that we received a contiguous subsequence of the states of the remote object. For each run of the test, n
+	// may be slightly different due to scheduling effects.
+	for i := 0; i < len(observedStates); i++ {
+		expectedState := observedStates[0] + uint64(i)
+		if observedStates[i] != expectedState {
+			return fmt.Errorf("expected observed state %v to be %v, got %v", i, expectedState, observedStates[i])
+		}
+	}
+	return nil
+}
+
+func serverProcess(ctx context.Context, srv *Conn, messages <-chan *Message, t *testing.T) error {
+	state := uint64(0)
+
+process:
+	for {
+		select {
+		case msg, ok := <-messages:
+			if !ok {
+				t.Log("Message channel closed")
+				// Message channel closed.
+				break process
+			}
+			if msg.IsValid() != nil {
+				t.Log("Got invalid message, discarding")
+				continue process
+			}
+			name := msg.Headers[FieldMember].value.(string)
+			ifname := msg.Headers[FieldInterface].value.(string)
+			if ifname == SCPPInterface && name == SCPPStateMethodName {
+				t.Logf("Processing reply to .State(), returning state = %v", state)
+				reply := new(Message)
+				reply.Type = TypeMethodReply
+				reply.serial = srv.getSerial()
+				reply.Headers = make(map[HeaderField]Variant)
+				reply.Headers[FieldDestination] = msg.Headers[FieldSender]
+				reply.Headers[FieldReplySerial] = MakeVariant(msg.serial)
+				reply.Body = make([]interface{}, 1)
+				reply.Body[0] = state
+				reply.Headers[FieldSignature] = MakeVariant(SignatureOf(reply.Body...))
+				srv.sendMessageAndIfClosed(reply, nil)
+			}
+		case <-ctx.Done():
+			t.Logf("Context cancelled, server emitted %v signals", state)
+			return nil
+		default:
+			state++
+			if err := srv.Emit(SCPPPath, SCPPInterface+"."+SCPPChangedSignalName, state); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 type server struct{}
 
 func (server) Double(i int64) (int64, *Error) {
diff --git a/sequence_test.go b/sequence_test.go
new file mode 100644
index 0000000..3482fea
--- /dev/null
+++ b/sequence_test.go
@@ -0,0 +1,16 @@
+package dbus
+
+import (
+	"testing"
+)
+
+func TestSequential(t *testing.T) {
+	gen := newSequenceGenerator()
+
+	for i := 1; i < 10; i++ {
+		sequence := gen.next()
+		if sequence != Sequence(i) {
+			t.Errorf("Got %v expected %v", sequence, i)
+		}
+	}
+}
diff --git a/sequential_handler_test.go b/sequential_handler_test.go
new file mode 100644
index 0000000..3adc52d
--- /dev/null
+++ b/sequential_handler_test.go
@@ -0,0 +1,258 @@
+package dbus
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+)
+
+// Verifies that no signals are dropped, even if there is not enough space
+// in the destination channel.
+func TestSequentialHandlerNoDrop(t *testing.T) {
+	t.Parallel()
+
+	handler := NewSequentialSignalHandler()
+
+	channel := make(chan *Signal, 2)
+	handler.(SignalRegistrar).AddSignal(channel)
+
+	writeSignals(handler, 1000)
+
+	if err := readSignals(t, channel, 1000); err != nil {
+		t.Error(err)
+	}
+}
+
+// Verifies that signals are written to the destination channel in the
+// order they are received, in a typical concurrent reader/writer scenario.
+func TestSequentialHandlerSequential(t *testing.T) {
+	t.Parallel()
+
+	handler := NewSequentialSignalHandler()
+
+	channel := make(chan *Signal, 10)
+	handler.(SignalRegistrar).AddSignal(channel)
+
+	done := make(chan struct{})
+
+	// Concurrently read and write signals
+	go func() {
+		if err := readSignals(t, channel, 1000); err != nil {
+			t.Error(err)
+		}
+		close(done)
+	}()
+	writeSignals(handler, 1000)
+	<-done
+}
+
+// Test that in the case of multiple destination channels, one channel
+// being blocked does not prevent the other channel receiving messages.
+func TestSequentialHandlerMultipleChannel(t *testing.T) {
+	t.Parallel()
+
+	handler := NewSequentialSignalHandler()
+
+	channelOne := make(chan *Signal)
+	handler.(SignalRegistrar).AddSignal(channelOne)
+
+	channelTwo := make(chan *Signal, 10)
+	handler.(SignalRegistrar).AddSignal(channelTwo)
+
+	writeSignals(handler, 1000)
+
+	if err := readSignals(t, channelTwo, 1000); err != nil {
+		t.Error(err)
+	}
+}
+
+// Test that removing one channel results in no more messages being
+// written to that channel.
+func TestSequentialHandler_RemoveOneChannelOfOne(t *testing.T) {
+	t.Parallel()
+	handler := NewSequentialSignalHandler()
+
+	channelOne := make(chan *Signal)
+	handler.(SignalRegistrar).AddSignal(channelOne)
+
+	writeSignals(handler, 1000)
+
+	handler.(SignalRegistrar).RemoveSignal(channelOne)
+
+	count, closed := countSignals(channelOne)
+	if count > 1 {
+		t.Error("handler continued writing to channel after removal")
+	}
+	if closed {
+		t.Error("handler closed channel on .RemoveChannel()")
+	}
+}
+
+// Test that removing one channel results in no more messages being
+// written to that channel, and the other channels are unaffected.
+func TestSequentialHandler_RemoveOneChannelOfMany(t *testing.T) {
+	t.Parallel()
+	handler := NewSequentialSignalHandler()
+
+	channelOne := make(chan *Signal)
+	handler.(SignalRegistrar).AddSignal(channelOne)
+
+	channelTwo := make(chan *Signal, 10)
+	handler.(SignalRegistrar).AddSignal(channelTwo)
+
+	channelThree := make(chan *Signal, 2)
+	handler.(SignalRegistrar).AddSignal(channelThree)
+
+	writeSignals(handler, 1000)
+
+	handler.(SignalRegistrar).RemoveSignal(channelTwo)
+	defer close(channelTwo)
+
+	count, closed := countSignals(channelTwo)
+	if count > 10 {
+		t.Error("handler continued writing to channel after removal")
+	}
+	if closed {
+		t.Error("handler closed channel on .RemoveChannel()")
+	}
+
+	// Check that closing channel two does not close channel one.
+	if err := readSignals(t, channelOne, 1000); err != nil {
+		t.Error(err)
+	}
+
+	// Check that closing channel two does not close channel three.
+	if err := readSignals(t, channelThree, 1000); err != nil {
+		t.Error(err)
+	}
+}
+
+// Test that Terminate() closes all channels that were attached at the time.
+func TestSequentialHandler_TerminateClosesAllChannels(t *testing.T) {
+	t.Parallel()
+	handler := NewSequentialSignalHandler()
+
+	channelOne := make(chan *Signal)
+	handler.(SignalRegistrar).AddSignal(channelOne)
+
+	channelTwo := make(chan *Signal, 10)
+	handler.(SignalRegistrar).AddSignal(channelTwo)
+
+	writeSignals(handler, 1000)
+
+	handler.(Terminator).Terminate()
+
+	count, closed := countSignals(channelOne)
+	if count > 1 {
+		t.Errorf("handler continued writing to channel after termination; read %v signals", count)
+	}
+	if !closed {
+		t.Error("handler failed to close channel on .Terminate()")
+	}
+
+	count, closed = countSignals(channelTwo)
+	if count > 10 {
+		t.Errorf("handler continued writing to channel after termination; read %v signals", count)
+	}
+	if !closed {
+		t.Error("handler failed to close channel on .Terminate()")
+	}
+}
+
+// Verifies that after termination, the handler does not process any further signals.
+func TestSequentialHandler_TerminateTerminates(t *testing.T) {
+	t.Parallel()
+	handler := NewSequentialSignalHandler()
+	handler.(Terminator).Terminate()
+
+	channelOne := make(chan *Signal)
+	handler.(SignalRegistrar).AddSignal(channelOne)
+
+	writeSignals(handler, 10)
+
+	count, _ := countSignals(channelOne)
+	if count > 0 {
+		t.Errorf("handler continued operating after termination; read %v signals", count)
+	}
+}
+
+// Verifies calling .Terminate() more than once is equivalent to calling it just once.
+func TestSequentialHandler_TerminateIdemopotent(t *testing.T) {
+	t.Parallel()
+	handler := NewSequentialSignalHandler()
+	handler.(Terminator).Terminate()
+	handler.(Terminator).Terminate()
+
+	channelOne := make(chan *Signal)
+	handler.(SignalRegistrar).AddSignal(channelOne)
+	writeSignals(handler, 10)
+
+	count, _ := countSignals(channelOne)
+	if count > 0 {
+		t.Errorf("handler continued operating after termination; read %v signals", count)
+	}
+}
+
+// Verifies calling RemoveSignal after Terminate() does not cause any unusual
+// behaviour (panics, etc.).
+func TestSequentialHandler_RemoveAfterTerminate(t *testing.T) {
+	t.Parallel()
+	handler := NewSequentialSignalHandler()
+	handler.(Terminator).Terminate()
+	handler.(Terminator).Terminate()
+
+	channelOne := make(chan *Signal)
+	handler.(SignalRegistrar).AddSignal(channelOne)
+	handler.(SignalRegistrar).RemoveSignal(channelOne)
+	writeSignals(handler, 10)
+
+	count, _ := countSignals(channelOne)
+	if count > 0 {
+		t.Errorf("handler continued operating after termination; read %v signals", count)
+	}
+}
+
+func writeSignals(handler SignalHandler, count int) {
+	for i := 1; i <= count; i++ {
+		signal := &Signal{Sequence: Sequence(i)}
+		handler.DeliverSignal("iface", "name", signal)
+	}
+}
+
+func readSignals(t *testing.T, channel <-chan *Signal, count int) error {
+	// Overly generous timeout
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	defer cancel()
+	for i := 1; i <= count; i++ {
+		select {
+		case signal := <-channel:
+			if signal.Sequence != Sequence(i) {
+				return fmt.Errorf("Received signal out of order. Expected %v, got %v", i, signal.Sequence)
+			}
+		case <-ctx.Done():
+			return errors.New("Timeout occured before all messages received")
+		}
+	}
+	return nil
+}
+
+func countSignals(channel <-chan *Signal) (count int, closed bool) {
+	count = 0
+	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*100)
+	defer cancel()
+	for {
+		select {
+		case _, ok := <-channel:
+			if ok {
+				count++
+			} else {
+				// Channel closed
+				return count, true
+			}
+		case <-ctx.Done():
+			return count, false
+		}
+	}
+}