Repository: creachadair/jrpc2
Pull request #64 (2021-12-03 23:46:45)

Add goroutine leak checking to unit tests.
Use the https://godoc.org/github.com/fortytw2/leaktest package to test for
leaked goroutines at exit in tests that use them.


Patch:
diff --git a/go.mod b/go.mod
index a999c4c..9a48280 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,7 @@
 module github.com/creachadair/jrpc2
 
 require (
+	github.com/fortytw2/leaktest v1.3.0
 	github.com/google/go-cmp v0.5.6
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
diff --git a/go.sum b/go.sum
index 632f99d..e09ae32 100644
--- a/go.sum
+++ b/go.sum
@@ -1,3 +1,5 @@
+github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=
+github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
 github.com/google/go-cmp v0.5.6 h1:BKbKCqvP6I+rmFHt06ZmyQtvB8xAkWdhFyr0ZUNZcxQ=
 github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=


Test patch:
diff --git a/channel/channel_test.go b/channel/channel_test.go
index 6a0a77c..d540a7a 100644
--- a/channel/channel_test.go
+++ b/channel/channel_test.go
@@ -8,6 +8,8 @@ import (
 	"strings"
 	"sync"
 	"testing"
+
+	"github.com/fortytw2/leaktest"
 )
 
 // newPipe creates a pair of connected in-memory channels using the specified
@@ -22,6 +24,8 @@ func newPipe(framing Framing) (client, server Channel) {
 }
 
 func testSendRecv(t *testing.T, s, r Channel, msg string) {
+	defer leaktest.Check(t)()
+
 	var wg sync.WaitGroup
 	var sendErr, recvErr error
 	var data []byte
diff --git a/internal_test.go b/internal_test.go
index ac8af67..8daa00d 100644
--- a/internal_test.go
+++ b/internal_test.go
@@ -14,6 +14,7 @@ import (
 
 	"github.com/creachadair/jrpc2/channel"
 	"github.com/creachadair/jrpc2/code"
+	"github.com/fortytw2/leaktest"
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 )
@@ -152,6 +153,8 @@ func (h hmap) Names() []string                                 { return nil }
 // Verify that if the client context terminates during a request, the client
 // will terminate and report failure.
 func TestClient_contextCancellation(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	started := make(chan struct{})
 	stopped := make(chan struct{})
 	cpipe, spipe := channel.Direct()
@@ -205,6 +208,8 @@ func TestClient_contextCancellation(t *testing.T) {
 }
 
 func TestServer_specialMethods(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	s := NewServer(hmap{
 		"rpc.nonesuch": methodFunc(func(context.Context, *Request) (interface{}, error) {
 			return "OK", nil
@@ -227,6 +232,8 @@ func TestServer_specialMethods(t *testing.T) {
 // Verify that the option to remove the special behaviour of rpc.* methods can
 // be correctly disabled by the server options.
 func TestServer_disableBuiltinHook(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	s := NewServer(hmap{
 		"rpc.nonesuch": methodFunc(func(context.Context, *Request) (interface{}, error) {
 			return "OK", nil
@@ -251,6 +258,8 @@ func TestServer_disableBuiltinHook(t *testing.T) {
 // request. The Client never sends requests like that, but the server needs to
 // cope with it correctly.
 func TestBatchReply(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	cpipe, spipe := channel.Direct()
 	srv := NewServer(hmap{
 		"test": methodFunc(func(_ context.Context, req *Request) (interface{}, error) {
diff --git a/jhttp/jhttp_test.go b/jhttp/jhttp_test.go
index 8b5003b..2290885 100644
--- a/jhttp/jhttp_test.go
+++ b/jhttp/jhttp_test.go
@@ -16,6 +16,7 @@ import (
 	"github.com/creachadair/jrpc2"
 	"github.com/creachadair/jrpc2/handler"
 	"github.com/creachadair/jrpc2/jhttp"
+	"github.com/fortytw2/leaktest"
 )
 
 var testService = handler.Map{
@@ -35,6 +36,8 @@ func checkContext(ctx context.Context, _ string, p json.RawMessage) (json.RawMes
 }
 
 func TestBridge(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	// Set up a bridge with the test configuration.
 	b := jhttp.NewBridge(testService, &jhttp.BridgeOptions{
 		Client: &jrpc2.ClientOptions{EncodeContext: checkContext},
@@ -163,6 +166,8 @@ func TestBridge(t *testing.T) {
 
 // Verify that the content-type check hook works.
 func TestBridge_contentTypeCheck(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	b := jhttp.NewBridge(testService, &jhttp.BridgeOptions{
 		CheckContentType: func(ctype string) bool {
 			return ctype == "application/octet-stream"
@@ -197,6 +202,8 @@ func TestBridge_contentTypeCheck(t *testing.T) {
 
 // Verify that the content-type check hook works.
 func TestBridge_requestCheck(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	b := jhttp.NewBridge(testService, &jhttp.BridgeOptions{
 		CheckRequest: func(req *http.Request) error {
 			if req.Header.Get("x-test-header") == "fail" {
@@ -245,6 +252,8 @@ func TestBridge_requestCheck(t *testing.T) {
 }
 
 func TestChannel(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	b := jhttp.NewBridge(testService, nil)
 	defer checkClose(t, b)
 	hsrv := httptest.NewServer(b)
diff --git a/jrpc2_test.go b/jrpc2_test.go
index b40a21c..838cc15 100644
--- a/jrpc2_test.go
+++ b/jrpc2_test.go
@@ -19,6 +19,7 @@ import (
 	"github.com/creachadair/jrpc2/handler"
 	"github.com/creachadair/jrpc2/jctx"
 	"github.com/creachadair/jrpc2/server"
+	"github.com/fortytw2/leaktest"
 	"github.com/google/go-cmp/cmp"
 )
 
@@ -115,6 +116,8 @@ var callTests = []struct {
 }
 
 func TestServerInfo_methodNames(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	loc := server.NewLocal(handler.ServiceMap{
 		"Test": testService,
 	}, nil)
@@ -131,6 +134,8 @@ func TestServerInfo_methodNames(t *testing.T) {
 }
 
 func TestClient_Call(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	loc := server.NewLocal(handler.ServiceMap{
 		"Test": testService,
 	}, &server.LocalOptions{
@@ -162,6 +167,8 @@ func TestClient_Call(t *testing.T) {
 }
 
 func TestClient_CallResult(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	loc := server.NewLocal(handler.ServiceMap{
 		"Test": testService,
 	}, &server.LocalOptions{
@@ -185,6 +192,8 @@ func TestClient_CallResult(t *testing.T) {
 }
 
 func TestClient_Batch(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	loc := server.NewLocal(handler.ServiceMap{
 		"Test": testService,
 	}, &server.LocalOptions{
@@ -233,6 +242,8 @@ func TestClient_Batch(t *testing.T) {
 
 // Verify that notifications respect order of arrival.
 func TestServer_notificationOrder(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	var last int32
 
 	loc := server.NewLocal(handler.Map{
@@ -264,6 +275,8 @@ func TestServer_notificationOrder(t *testing.T) {
 // Verify that a method that returns only an error (no result payload) is set
 // up and handled correctly.
 func TestHandler_errorOnly(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	const errMessage = "not enough strings"
 	loc := server.NewLocal(handler.Map{
 		"ErrorOnly": handler.New(func(_ context.Context, ss []string) error {
@@ -307,6 +320,8 @@ func TestHandler_errorOnly(t *testing.T) {
 // Verify that a timeout set on the context is respected by the server and
 // propagates back to the client as an error.
 func TestServer_contextTimeout(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	loc := server.NewLocal(handler.Map{
 		"Stall": handler.New(func(ctx context.Context) (bool, error) {
 			t.Log("Stalling...")
@@ -338,6 +353,8 @@ func TestServer_contextTimeout(t *testing.T) {
 
 // Verify that stopping the server terminates in-flight requests.
 func TestServer_stopCancelsHandlers(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	started := make(chan struct{})
 	stopped := make(chan error, 1)
 	loc := server.NewLocal(handler.Map{
@@ -374,6 +391,8 @@ func TestServer_stopCancelsHandlers(t *testing.T) {
 
 // Test that a handler can cancel an in-flight request.
 func TestServer_CancelRequest(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	ready := make(chan struct{})
 	loc := server.NewLocal(handler.Map{
 		"Stall": handler.New(func(ctx context.Context) error {
@@ -425,6 +444,8 @@ func TestServer_CancelRequest(t *testing.T) {
 // Test that an error with data attached to it is correctly propagated back
 // from the server to the client, in a value of concrete type *Error.
 func TestError_withData(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	const errCode = -32000
 	const errData = `{"caroline":452}`
 	const errMessage = "error thingy"
@@ -477,6 +498,8 @@ func TestError_withData(t *testing.T) {
 
 // Test that a client correctly reports bad parameters.
 func TestClient_badCallParams(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	loc := server.NewLocal(handler.Map{
 		"Test": handler.New(func(_ context.Context, v interface{}) error {
 			return jrpc2.Errorf(129, "this should not be reached")
@@ -494,6 +517,8 @@ func TestClient_badCallParams(t *testing.T) {
 
 // Verify that metrics are correctly propagated to server info.
 func TestServer_serverInfoMetrics(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	loc := server.NewLocal(handler.Map{
 		"Metricize": handler.New(func(ctx context.Context) (bool, error) {
 			m := jrpc2.ServerFromContext(ctx).Metrics()
@@ -557,6 +582,8 @@ func TestServer_serverInfoMetrics(t *testing.T) {
 // elicit a correct response from the server. Here we simulate a "different"
 // client by writing requests directly into the channel.
 func TestServer_nonLibraryClient(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	srv, cli := channel.Direct()
 	s := jrpc2.NewServer(handler.Map{
 		"X": testOK,
@@ -672,6 +699,8 @@ func TestServer_nonLibraryClient(t *testing.T) {
 
 // Verify that server-side push notifications work.
 func TestServer_Notify(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	// Set up a server and client with server-side notification support.  Here
 	// we're just capturing the name of the notification method, as a sign we
 	// got the right thing.
@@ -723,6 +752,8 @@ func TestServer_Notify(t *testing.T) {
 
 // Verify that server-side callbacks can time out.
 func TestServer_callbackTimeout(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	loc := server.NewLocal(handler.Map{
 		"Test": handler.New(func(ctx context.Context) error {
 			tctx, cancel := context.WithTimeout(ctx, 5*time.Millisecond)
@@ -752,6 +783,8 @@ func TestServer_callbackTimeout(t *testing.T) {
 
 // Verify that server-side callbacks work.
 func TestServer_Callback(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	loc := server.NewLocal(handler.Map{
 		"CallMeMaybe": handler.New(func(ctx context.Context) error {
 			if _, err := jrpc2.ServerFromContext(ctx).Callback(ctx, "succeed", nil); err != nil {
@@ -794,6 +827,8 @@ func TestServer_Callback(t *testing.T) {
 
 // Verify that a server push after the client closes does not trigger a panic.
 func TestServer_pushAfterClose(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	loc := server.NewLocal(make(handler.Map), &server.LocalOptions{
 		Server: &jrpc2.ServerOptions{AllowPush: true},
 	})
@@ -809,6 +844,8 @@ func TestServer_pushAfterClose(t *testing.T) {
 
 // Verify that an OnCancel hook is called when expected.
 func TestClient_onCancelHook(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	hooked := make(chan struct{}) // closed when hook notification is finished
 
 	loc := server.NewLocal(handler.Map{
@@ -867,6 +904,8 @@ func TestClient_onCancelHook(t *testing.T) {
 
 // Verify that client callback handlers are cancelled when the client stops.
 func TestClient_closeEndsCallbacks(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	ready := make(chan struct{})
 	loc := server.NewLocal(handler.Map{
 		"Test": handler.New(func(ctx context.Context) error {
@@ -906,6 +945,8 @@ func TestClient_closeEndsCallbacks(t *testing.T) {
 // Verify that it is possible for multiple callback handlers to execute
 // concurrently.
 func TestClient_concurrentCallbacks(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	ready1 := make(chan struct{})
 	ready2 := make(chan struct{})
 	release := make(chan struct{})
@@ -963,6 +1004,7 @@ func TestClient_concurrentCallbacks(t *testing.T) {
 			}),
 		},
 	})
+	defer loc.Close()
 
 	var got []string
 	if err := loc.Client.CallResult(context.Background(), "Test", nil, &got); err != nil {
@@ -976,6 +1018,8 @@ func TestClient_concurrentCallbacks(t *testing.T) {
 
 // Verify that a callback can successfully call "up" into the server.
 func TestClient_callbackUpCall(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	const pingMessage = "kittens!"
 
 	var probe string
@@ -1012,6 +1056,8 @@ func TestClient_callbackUpCall(t *testing.T) {
 
 // Verify that the context encoding/decoding hooks work.
 func TestContextPlumbing(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	want := time.Now().Add(10 * time.Second)
 	ctx, cancel := context.WithDeadline(context.Background(), want)
 	defer cancel()
@@ -1041,6 +1087,8 @@ func TestContextPlumbing(t *testing.T) {
 // Verify that calling a wrapped method which takes no parameters, but in which
 // the caller provided parameters, will correctly report an error.
 func TestHandler_noParams(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	loc := server.NewLocal(handler.Map{"Test": testOK}, nil)
 	defer loc.Close()
 
@@ -1054,6 +1102,8 @@ func TestHandler_noParams(t *testing.T) {
 
 // Verify that the rpc.serverInfo handler and client wrapper work together.
 func TestRPCServerInfo(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	loc := server.NewLocal(handler.Map{"Test": testOK}, nil)
 	defer loc.Close()
 
@@ -1101,6 +1151,8 @@ func TestNetwork(t *testing.T) {
 
 // Verify that the context passed to an assigner has the correct structure.
 func TestHandler_assignContext(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	loc := server.NewLocal(assignFunc(func(ctx context.Context, method string) jrpc2.Handler {
 		req := jrpc2.InboundRequest(ctx)
 		if req == nil {
@@ -1129,6 +1181,8 @@ func (a assignFunc) Assign(ctx context.Context, m string) jrpc2.Handler { return
 func (assignFunc) Names() []string                                      { return nil }
 
 func TestServer_WaitStatus(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	check := func(t *testing.T, stat jrpc2.ServerStatus, closed, stopped bool, wantErr error) {
 		t.Helper()
 		if got, want := stat.Success(), wantErr == nil; got != want {
@@ -1174,6 +1228,8 @@ func (b buggyChannel) Recv() ([]byte, error) { return []byte(b.data), b.err }
 func (buggyChannel) Close() error            { return nil }
 
 func TestRequest_strictFields(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	type other struct {
 		C bool `json:"charlie"`
 	}
@@ -1232,6 +1288,8 @@ func TestRequest_strictFields(t *testing.T) {
 }
 
 func TestServerFromContext(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	var got *jrpc2.Server
 	loc := server.NewLocal(handler.Map{
 		"Test": handler.New(func(ctx context.Context) error {
@@ -1251,6 +1309,8 @@ func TestServerFromContext(t *testing.T) {
 }
 
 func TestServer_newContext(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	// Prepare a context with a test value attached to it, that the handler can
 	// extract to verify that the base context was plumbed in correctly.
 	type ctxKey string
diff --git a/server/local_test.go b/server/local_test.go
index 21c79fe..c34788a 100644
--- a/server/local_test.go
+++ b/server/local_test.go
@@ -11,6 +11,7 @@ import (
 	"github.com/creachadair/jrpc2"
 	"github.com/creachadair/jrpc2/handler"
 	"github.com/creachadair/jrpc2/server"
+	"github.com/fortytw2/leaktest"
 )
 
 var doDebug = flag.Bool("debug", false, "Enable server and client debugging logs")
@@ -26,6 +27,8 @@ func testOpts(t *testing.T) *server.LocalOptions {
 }
 
 func TestLocal(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	loc := server.NewLocal(make(handler.Map), testOpts(t))
 	ctx := context.Background()
 	si, err := jrpc2.RPCServerInfo(ctx, loc.Client)
@@ -49,6 +52,8 @@ func TestLocal(t *testing.T) {
 
 // Test that concurrent callers to a local service do not deadlock.
 func TestLocalConcurrent(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	loc := server.NewLocal(handler.Map{
 		"Test": handler.New(func(context.Context) error { return nil }),
 	}, testOpts(t))
diff --git a/server/loop_test.go b/server/loop_test.go
index 009e8ef..8ac68b6 100644
--- a/server/loop_test.go
+++ b/server/loop_test.go
@@ -13,6 +13,7 @@ import (
 	"github.com/creachadair/jrpc2"
 	"github.com/creachadair/jrpc2/channel"
 	"github.com/creachadair/jrpc2/handler"
+	"github.com/fortytw2/leaktest"
 )
 
 var newChan = channel.Line
@@ -101,6 +102,8 @@ func mustServe(t *testing.T, lst net.Listener, newService func() Service) <-chan
 
 // Test that sequential clients against the same server work sanely.
 func TestSeq(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	lst := mustListen(t)
 	addr := lst.Addr().String()
 	sc := mustServe(t, lst, testService)
@@ -121,6 +124,8 @@ func TestSeq(t *testing.T) {
 
 // Test that concurrent clients against the same server work sanely.
 func TestLoop(t *testing.T) {
+	defer leaktest.Check(t)()
+
 	tests := []struct {
 		desc string
 		cons func() Service