Repository: creachadair/jrpc2 (go)

Issue:
Support sequential request processing or selective request ordering
Asynchronous processing of requests is useful from performance perspective. Unfortunately there are scenarios in LSP where ordering matters.

One example is when the client sends a number of partial text edits which need to be applied in that particular order.

As per LSP it is necessary to process any `textDocument/didChange` requests in the same order they were received:

https://microsoft.github.io/language-server-protocol/specifications/specification-current/#textDocument_didChange

```ts
        /*
	 * To mirror the content of a document using change events use the following approach:
	 * - start with the same initial content
	 * - apply the 'textDocument/didChange' notifications in the order you receive them.
	 * - apply the `TextDocumentContentChangeEvent`s in a single notification in the order
	 *   you receive them.
	 */
```

In theory the incoming requests could be ordered by IDs, but:

1. IDs are not even guaranteed to be numbers in LSP and even when they are the numbers may not have any meaning
2. Even though clients seem to be sending IDs in `textDocument/didChange` requests, the LSP technically documents this as a notification method - i.e. request should not contain ID and should not be responded to

In some cases it may be possible to order requests by version ID, but the sequence may not be complete, so there is no point in queuing the requests and waiting if a particular version arrives.

https://microsoft.github.io/language-server-protocol/specifications/specification-current/#versionedTextDocumentIdentifier

> The version number of a document will increase after each change, including
> undo/redo. The number doesn't need to be consecutive.

--- 

`gopls` (Go language server) seems to be implementing purely sequential server, which I don't is strictly necessary, but it probably reduces the complexity quite a bit which may be preferable. i.e. It is IMO sensible to temporarily compromise the performance in favour of getting the overall implementation right.

Lastly the LSP spec says the following about ordering:

> **Request, Notification and Response ordering**
> 
> Responses to requests should be sent in roughly the same order as the requests appear on the server or client side. So for example if a server receives a textDocument/completion request and then a textDocument/signatureHelp request it will usually first return the response for the textDocument/completion and then the response for textDocument/signatureHelp.
> 
> However, the server may decide to use a parallel execution strategy and may wish to return responses in a different order than the requests were received. The server may do so as long as this reordering doesn’t affect the correctness of the responses. For example, reordering the result of textDocument/completion and textDocument/signatureHelp is allowed, as these each of these requests usually won’t affect the output of the other. On the other hand, the server most likely should not reorder textDocument/definition and textDocument/rename requests, since the executing the latter may affect the result of the former.

Pull request:
Add a barrier constraining execution order for notifications.
An attempt to address #20.

When issuing a batch of requests, keep track of the number that are valid notifications, and do not issue to any handler for the batch until all notifications received in previous batches have completed.

This ensures a sensible partial order on notifications. Multiple notifications in a single batch can still be executed in parallel, but notifications across batches are no longer concurrent.

Before merging:

- [x] Verify that this change fixes #20 (see hashicorp/terraform-ls#161 (comment))
- [x] Update the documentation for this new behaviour.

Patch:
diff --git a/base.go b/base.go
index 39d1cc0..4749c7a 100644
--- a/base.go
+++ b/base.go
@@ -322,6 +322,9 @@ func (j *jmessage) parseJSON(data []byte) error {
 // isRequestOrNotification reports whether j is a request or notification.
 func (j *jmessage) isRequestOrNotification() bool { return j.E == nil && j.R == nil && j.M != "" }
 
+// isNotification reports whether j is a notification
+func (j *jmessage) isNotification() bool { return j.isRequestOrNotification() && fixID(j.ID) == nil }
+
 type jerror struct {
 	C int32           `json:"code"`
 	M string          `json:"message,omitempty"`
diff --git a/doc.go b/doc.go
index 4afdc8c..197507f 100644
--- a/doc.go
+++ b/doc.go
@@ -185,6 +185,21 @@ name on the first period ("."), and you may nest ServiceMaps more deeply if you
 require a more complex hierarchy.
 
 
+Concurrency
+
+A Server processes requests concurrently, up to the Concurrency limit given in
+its ServerOptions. Two requests (calls or notifications) are concurrent if they
+arrive as part of the same batch. In addition, two calls are concurrent if the
+time intervals between the arrival of the request objects and delivery of the
+response objects overlap.
+
+The server may issue concurrent requests to their handlers in any order.
+Otherwise, requests are processed in order of arrival. Notifications, in
+particular, can only be concurrent with other notifications in the same batch.
+This ensures a client that sends a notification can be sure its notification
+was fully processed before any subsequent calls are issued.
+
+
 Non-Standard Extension Methods
 
 By default a jrpc2.Server exports the following built-in non-standard extension
diff --git a/opts.go b/opts.go
index 9d7e758..29ec43f 100644
--- a/opts.go
+++ b/opts.go
@@ -38,7 +38,8 @@ type ServerOptions struct {
 	DisableBuiltin bool
 
 	// Allows up to the specified number of goroutines to execute concurrently
-	// in request handlers. A value less than 1 uses runtime.NumCPU().
+	// in request handlers. A value less than 1 uses runtime.NumCPU().  Note
+	// that this setting does not constrain order of issue.
 	Concurrency int
 
 	// If set, this function is called with the method name and encoded request
diff --git a/server.go b/server.go
index fdd3a5c..ff824c2 100644
--- a/server.go
+++ b/server.go
@@ -37,6 +37,7 @@ type Server struct {
 
 	mu *sync.Mutex // protects the fields below
 
+	nbar sync.WaitGroup  // notification barrier (see the dispatch method)
 	err  error           // error from a previous operation
 	work *sync.Cond      // for signaling message availability
 	inq  *list.List      // inbound requests awaiting processing
@@ -171,35 +172,50 @@ func (s *Server) nextRequest() (func() error, error) {
 // dispatch constructs a function that invokes each of the specified tasks.
 // The caller must hold s.mu when calling dispatch, but the returned function
 // should be executed outside the lock to wait for the handlers to return.
+//
+// dispatch blocks until any notification received prior to this batch has
+// completed, to ensure that notifications are processed in a partial order
+// that respects order of receipt. Notifications within a batch are handled
+// concurrently.
 func (s *Server) dispatch(next jmessages, ch channel.Sender) func() error {
 	// Resolve all the task handlers or record errors.
 	start := time.Now()
 	tasks := s.checkAndAssign(next)
+	last := len(tasks) - 1
 
-	n := len(tasks)
-	last := tasks[n-1]
+	// s.nbar counts the number of notifications that have been issued and are
+	// not yet complete. Before issuing any tasks in this batch, wait for all
+	// such notifications to complete, and update the barrier with the number in
+	// this batch. This update must happen under s.mu, so that multiple batches
+	// do not race on Wait/Add.
+	s.nbar.Wait()
+	s.nbar.Add(tasks.numValidNotifications())
 
-	// If there are multiple tasks, start goroutines to handle all but the
-	// last. The last is handled directly, to avoid spinning up a separate
-	// goroutine for a single task.
-	var wg sync.WaitGroup
-	for _, t := range tasks[:n-1] {
-		if t.err != nil {
-			continue // nothing to do here; this task has already failed
+	return func() error {
+		var wg sync.WaitGroup
+		for i, t := range tasks {
+			if t.err != nil {
+				continue // nothing to do here; this task has already failed
+			}
+			t := t
+
+			wg.Add(1)
+			run := func() {
+				defer wg.Done()
+				if t.hreq.IsNotification() {
+					defer s.nbar.Done()
+				}
+				t.val, t.err = s.invoke(t.ctx, t.m, t.hreq)
+			}
+			if i < last {
+				go run()
+			} else {
+				run()
+			}
 		}
-		wg.Add(1)
-		go func(t *task) {
-			defer wg.Done()
-			t.val, t.err = s.invoke(t.ctx, t.m, t.hreq)
-		}(t)
-	}
 
-	// Wait for all the handlers to return, then deliver any responses.
-	return func() error {
+		// Wait for all the handlers to return, then deliver any responses.
 		wg.Wait()
-		if last.err == nil {
-			last.val, last.err = s.invoke(last.ctx, last.m, last.hreq)
-		}
 		return s.deliver(tasks.responses(s.rpcLog), ch, time.Since(start))
 	}
 }
@@ -421,7 +437,7 @@ func (s *Server) stop(err error) {
 	var keep jmessages
 	for cur := s.inq.Front(); cur != nil; cur = s.inq.Front() {
 		for _, req := range cur.Value.(jmessages) {
-			if req.ID == nil {
+			if req.isNotification() {
 				keep = append(keep, req)
 				s.log("Retaining notification %p", req)
 			} else {
@@ -613,3 +629,14 @@ func (ts tasks) responses(rpcLog RPCLogger) jmessages {
 	}
 	return rsps
 }
+
+// numValidNotifications reports the number of elements in ts that are
+// syntactically valid notifications.
+func (ts tasks) numValidNotifications() (n int) {
+	for _, t := range ts {
+		if t.err == nil && t.hreq.IsNotification() {
+			n++
+		}
+	}
+	return
+}


Test patch:
diff --git a/jrpc2_test.go b/jrpc2_test.go
index 58f3fb7..73f465b 100644
--- a/jrpc2_test.go
+++ b/jrpc2_test.go
@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -220,6 +221,36 @@ func TestBatch(t *testing.T) {
 	}
 }
 
+// Verify that notifications respect order of arrival.
+func TestNotificationOrder(t *testing.T) {
+	var last int32
+
+	loc := server.NewLocal(handler.Map{
+		"Test": handler.New(func(_ context.Context, req *jrpc2.Request) error {
+			var seq int32
+			if err := req.UnmarshalParams(&handler.Args{&seq}); err != nil {
+				t.Errorf("Invalid test parameters: %v", err)
+				return err
+			}
+			if old := atomic.SwapInt32(&last, seq); old != seq-1 {
+				t.Errorf("Request out of sequence at #%d: got %d, want %d", seq, old, seq-1)
+			}
+			return nil
+		}),
+	}, &server.LocalOptions{
+		Server: &jrpc2.ServerOptions{Concurrency: 16},
+	})
+
+	for i := 1; i < 10; i++ {
+		if err := loc.Client.Notify(context.Background(), "Test", []int{i}); err != nil {
+			t.Errorf("Test notification failed: %v", err)
+		}
+	}
+	if err := loc.Close(); err != nil {
+		t.Logf("Warning: error at server exit: %v", err)
+	}
+}
+
 // Verify that a method that returns only an error (no result payload) is set
 // up and handled correctly.
 func TestErrorOnly(t *testing.T) {