Repository: creachadair/jrpc2
Pull request #77 (2022-02-04 17:55:55)

server: handle duplicate request IDs within a single batch
Previously, if two or more calls arrived in a single batch with the same request ID, the server would process the first one and report the rest as having duplicate IDs.

This is subtly wrong, in that the client has no way of knowing which of the duplicates should get the successful call, and which should get the errors. Although a client sending such requests is already wrong, this makes the error harder to debug.

Instead, check for duplicates within the batch during assignment, and fail all the colliding requests.

- Add a regression test for duplicate requests in a single batch.
- Add a shared common error for InvalidParams.
- Add a shared error base for NoSuchMethod.



Patch:
diff --git a/base.go b/base.go
index 148ef41..d99d280 100644
--- a/base.go
+++ b/base.go
@@ -96,12 +96,12 @@ func (r *Request) UnmarshalParams(v interface{}) error {
 		dec := json.NewDecoder(bytes.NewReader(r.params))
 		dec.DisallowUnknownFields()
 		if err := dec.Decode(v); err != nil {
-			return Errorf(code.InvalidParams, "invalid parameters: %v", err.Error())
+			return errInvalidParams.WithData(err.Error())
 		}
 		return nil
 	}
 	if err := json.Unmarshal(r.params, v); err != nil {
-		return Errorf(code.InvalidParams, "invalid parameters: %v", err.Error())
+		return errInvalidParams.WithData(err.Error())
 	}
 	return nil
 }
diff --git a/error.go b/error.go
index 4bc81ef..c7dc9ed 100644
--- a/error.go
+++ b/error.go
@@ -46,12 +46,21 @@ var errClientStopped = errors.New("the client has been stopped")
 // errEmptyMethod is the error reported for an empty request method name.
 var errEmptyMethod = &Error{Code: code.InvalidRequest, Message: "empty method name"}
 
+// errNoSuchMethod is the error reported for an unknown method name.
+var errNoSuchMethod = &Error{Code: code.MethodNotFound, Message: "no such method"}
+
+// errDuplicateID is the error reported for a duplicated request ID.
+var errDuplicateID = &Error{Code: code.InvalidRequest, Message: "duplicate request ID"}
+
 // errInvalidRequest is the error reported for an invalid request object or batch.
 var errInvalidRequest = &Error{Code: code.ParseError, Message: "invalid request value"}
 
 // errEmptyBatch is the error reported for an empty request batch.
 var errEmptyBatch = &Error{Code: code.InvalidRequest, Message: "empty request batch"}
 
+// errInvalidParams is the error reported for invalid request parameters.
+var errInvalidParams = &Error{Code: code.InvalidParams, Message: "invalid parameters"}
+
 // ErrConnClosed is returned by a server's push-to-client methods if they are
 // called after the client connection is closed.
 var ErrConnClosed = errors.New("client connection is closed")
diff --git a/server.go b/server.go
index cca0e84..735557c 100644
--- a/server.go
+++ b/server.go
@@ -278,16 +278,15 @@ func (s *Server) deliver(rsps jmessages, ch sender, elapsed time.Duration) error
 // records errors for them as appropriate. The caller must hold s.mu.
 func (s *Server) checkAndAssign(next jmessages) tasks {
 	var ts tasks
+	var ids []string
+	dup := make(map[string]*task) // :: id ⇒ first task in batch with id
+
+	// Phase 1: Filter out responses from push calls and check for duplicate
+	// request ID.s
 	for _, req := range next {
 		fid := fixID(req.ID)
-		t := &task{
-			hreq:  &Request{id: fid, method: req.M, params: req.P},
-			batch: req.batch,
-		}
 		id := string(fid)
-		if req.err != nil {
-			t.err = req.err // deferred validation error
-		} else if !req.isRequestOrNotification() && s.call[id] != nil {
+		if !req.isRequestOrNotification() && s.call[id] != nil {
 			// This is a result or error for a pending push-call.
 			//
 			// N.B. It is important to check for this before checking for
@@ -296,24 +295,51 @@ func (s *Server) checkAndAssign(next jmessages) tasks {
 			delete(s.call, id)
 			rsp.ch <- req
 			continue // don't send a reply for this
-		} else if id != "" && s.used[id] != nil {
-			t.err = Errorf(code.InvalidRequest, "duplicate request id %q", id)
+		} else if req.err != nil {
+			// keep the existing error
 		} else if !s.versionOK(req.V) {
-			t.err = ErrInvalidVersion
-		} else if req.M == "" {
+			req.err = ErrInvalidVersion
+		}
+
+		t := &task{
+			hreq:  &Request{id: fid, method: req.M, params: req.P},
+			batch: req.batch,
+			err:   req.err,
+		}
+		if old := dup[id]; old != nil {
+			// A previous task already used this ID, fail both.
+			old.err = errDuplicateID.WithData(id)
+			t.err = old.err
+		} else if id != "" && s.used[id] != nil {
+			// A task from a previous batch already used this ID, fail this one.
+			t.err = errDuplicateID.WithData(id)
+		} else if id != "" {
+			// This is the first task with this ID in the batch.
+			dup[id] = t
+		}
+		ts = append(ts, t)
+		ids = append(ids, id)
+	}
+
+	// Phase 2: Assign method handlers and set up contexts.
+	for i, t := range ts {
+		id := ids[i]
+		if t.err != nil {
+			// deferred validation error
+		} else if t.hreq.method == "" {
 			t.err = errEmptyMethod
 		} else if s.setContext(t, id) {
-			t.m = s.assign(t.ctx, req.M)
+			t.m = s.assign(t.ctx, t.hreq.method)
 			if t.m == nil {
-				t.err = Errorf(code.MethodNotFound, "no such method %q", req.M)
+				t.err = errNoSuchMethod.WithData(t.hreq.method)
 			}
 		}
 
 		if t.err != nil {
-			s.log("Request check error for %q (params %q): %v", req.M, string(req.P), t.err)
+			s.log("Request check error for %q (params %q): %v",
+				t.hreq.method, string(t.hreq.params), t.err)
 			s.metrics.Count("rpc.errors", 1)
 		}
-		ts = append(ts, t)
 	}
 	return ts
 }


Test patch:
diff --git a/jrpc2_test.go b/jrpc2_test.go
index 651baf9..b2deaa8 100644
--- a/jrpc2_test.go
+++ b/jrpc2_test.go
@@ -619,7 +619,7 @@ func TestServer_nonLibraryClient(t *testing.T) {
 
 		// The method specified doesn't exist.
 		{`{"jsonrpc":"2.0", "id": 3, "method": "NoneSuch"}`,
-			`{"jsonrpc":"2.0","id":3,"error":{"code":-32601,"message":"no such method \"NoneSuch\""}}`},
+			`{"jsonrpc":"2.0","id":3,"error":{"code":-32601,"message":"no such method","data":"NoneSuch"}}`},
 
 		// The parameters are of the wrong form.
 		{`{"jsonrpc":"2.0", "id": 4, "method": "X", "params": "bogus"}`,
@@ -654,7 +654,7 @@ func TestServer_nonLibraryClient(t *testing.T) {
 
 		// A batch of invalid requests returns a batch of errors.
 		{`[{"jsonrpc": "2.0", "id": 6, "method":"bogus"}]`,
-			`[{"jsonrpc":"2.0","id":6,"error":{"code":-32601,"message":"no such method \"bogus\""}}]`},
+			`[{"jsonrpc":"2.0","id":6,"error":{"code":-32601,"message":"no such method","data":"bogus"}}]`},
 
 		// Batch requests return batch responses, even for a singleton.
 		{`[{"jsonrpc": "2.0", "id": 7, "method": "X"}]`, `[{"jsonrpc":"2.0","id":7,"result":"OK"}]`},
diff --git a/regression_test.go b/regression_test.go
index dea451b..5f917f6 100644
--- a/regression_test.go
+++ b/regression_test.go
@@ -13,6 +13,7 @@ import (
 	"github.com/creachadair/jrpc2/handler"
 	"github.com/creachadair/jrpc2/server"
 	"github.com/fortytw2/leaktest"
+	"github.com/google/go-cmp/cmp"
 )
 
 // Verify that a notification handler will not deadlock with the dispatcher on
@@ -141,7 +142,7 @@ func TestDuplicateIDCancellation(t *testing.T) {
 
 	// Send the duplicate, which should report an error.
 	send(duplicateReq)
-	expect(`{"jsonrpc":"2.0","id":1,"error":{"code":-32600,"message":"duplicate request id \"1\""}}`)
+	expect(`{"jsonrpc":"2.0","id":1,"error":{"code":-32600,"message":"duplicate request ID","data":"1"}}`)
 
 	// Unblock the handler, which should now complete. If the duplicate request
 	// caused the handler to cancel, it will have logged an error to fail the test.
@@ -151,3 +152,43 @@ func TestDuplicateIDCancellation(t *testing.T) {
 	cch.Close()
 	srv.Wait()
 }
+
+func TestCheckBatchDuplicateID(t *testing.T) {
+	defer leaktest.Check(t)()
+
+	srv, cli := channel.Direct()
+	s := jrpc2.NewServer(handler.Map{
+		"Test": testOK,
+	}, nil).Start(srv)
+	defer func() {
+		cli.Close()
+		if err := s.Wait(); err != nil {
+			t.Errorf("Server wait: unexpected error: %v", err)
+		}
+	}()
+
+	// A batch of requests containing two calls with the same ID.
+	const input = `[
+  {"jsonrpc": "2.0", "id": 1, "method": "Test"},
+  {"jsonrpc": "2.0", "id": 1, "method": "Test"},
+  {"jsonrpc": "2.0", "id": 2, "method": "Test"}
+]
+`
+	const errorReply = `{` +
+		`"jsonrpc":"2.0",` +
+		`"id":1,` +
+		`"error":{"code":-32600,"message":"duplicate request ID","data":"1"}` +
+		`}`
+	const want = `[` + errorReply + `,` + errorReply + `,` + `{"jsonrpc":"2.0","id":2,"result":"OK"}]`
+
+	if err := cli.Send([]byte(input)); err != nil {
+		t.Fatalf("Send %d bytes failed: %v", len(input), err)
+	}
+	rsp, err := cli.Recv()
+	if err != nil {
+		t.Fatalf("Recv failed: %v", err)
+	}
+	if diff := cmp.Diff(want, string(rsp)); diff != "" {
+		t.Errorf("Server response: (-want, +got)\n%s", diff)
+	}
+}