internal/jsonrpc2: change to a more synchronous dispatch model

Delivering each message in a go routine turned out to be problematic, there are some messages
that must be fully processed before later messages are started, and there was no way to guarantee that.
We now push concurrence handling up to the higher level, this has the disadvantage of not being able to guarantee
we respond to call messages correctly, but its a small price to pay.
The LSP currently processes each message fully in order blocking the handler, while we still work on basic
functionality.

Change-Id: If0648c77713ddbe4fed69da97a57696f433b8002
Reviewed-on: https://go-review.googlesource.com/c/149497
Run-TryBot: Ian Cottrell <iancottrell@google.com>
TryBot-Result: Gobot Gobot <gobot@golang.org>
Reviewed-by: Rebecca Stambler <rstambler@golang.org>

internal/jsonrpc2/jsonrpc2.go

@@ -29,15 +29,15 @@ type Conn struct {
 	pendingMu  sync.Mutex // protects the pending map
 	pending    map[ID]chan *Response
 	handlingMu sync.Mutex // protects the handling map
-	handling   map[ID]context.CancelFunc
+	handling   map[ID]handling
 }
 
 // Handler is an option you can pass to NewConn to handle incoming requests.
-// If the request returns true from IsNotify then the Handler should not return a
+// If the request returns false from IsNotify then the Handler must eventually
-// result or error, otherwise it should handle the Request and return either
+// call Reply on the Conn with the supplied request.
-// an encoded result, or an error.
+// Handlers are called synchronously, they should pass the work off to a go
-// Handlers must be concurrency-safe.
+// routine if they are going to take a long time.
-type Handler = func(context.Context, *Conn, *Request) (interface{}, *Error)
+type Handler func(context.Context, *Conn, *Request)
 
 // Canceler is an option you can pass to NewConn which is invoked for
 // cancelled outgoing requests.
@@ -46,7 +46,7 @@ type Handler = func(context.Context, *Conn, *Request) (interface{}, *Error)
 // It is okay to use the connection to send notifications, but the context will
 // be in the cancelled state, so you must do it with the background context
 // instead.
-type Canceler = func(context.Context, *Conn, *Request)
+type Canceler func(context.Context, *Conn, *Request)
 
 // NewErrorf builds a Error struct for the suppied message and code.
 // If args is not empty, message and args will be passed to Sprintf.
@@ -64,7 +64,7 @@ func NewConn(ctx context.Context, s Stream, options ...interface{}) *Conn {
 		stream:   s,
 		done:     make(chan struct{}),
 		pending:  make(map[ID]chan *Response),
-		handling: make(map[ID]context.CancelFunc),
+		handling: make(map[ID]handling),
 	}
 	for _, opt := range options {
 		switch opt := opt.(type) {
@@ -89,8 +89,10 @@ func NewConn(ctx context.Context, s Stream, options ...interface{}) *Conn {
 	}
 	if conn.handle == nil {
 		// the default handler reports a method error
-		conn.handle = func(ctx context.Context, c *Conn, r *Request) (interface{}, *Error) {
+		conn.handle = func(ctx context.Context, c *Conn, r *Request) {
-			return nil, NewErrorf(CodeMethodNotFound, "method %q not found", r.Method)
+			if r.IsNotify() {
+				c.Reply(ctx, r, nil, NewErrorf(CodeMethodNotFound, "method %q not found", r.Method))
+			}
 		}
 	}
 	if conn.cancel == nil {
@@ -126,10 +128,10 @@ func (c *Conn) Wait(ctx context.Context) error {
 // to propagate the cancel.
 func (c *Conn) Cancel(id ID) {
 	c.handlingMu.Lock()
-	cancel := c.handling[id]
+	handling, found := c.handling[id]
 	c.handlingMu.Unlock()
-	if cancel != nil {
+	if found {
-		cancel()
+		handling.cancel()
 	}
 }
 
@@ -215,6 +217,59 @@ func (c *Conn) Call(ctx context.Context, method string, params, result interface
 	}
 }
 
+// Reply sends a reply to the given request.
+// It is an error to call this if request was not a call.
+// You must call this exactly once for any given request.
+// If err is set then result will be ignored.
+func (c *Conn) Reply(ctx context.Context, req *Request, result interface{}, err error) error {
+	if req.IsNotify() {
+		return fmt.Errorf("reply not invoked with a valid call")
+	}
+	c.handlingMu.Lock()
+	handling, found := c.handling[*req.ID]
+	if found {
+		delete(c.handling, *req.ID)
+	}
+	c.handlingMu.Unlock()
+	if !found {
+		return fmt.Errorf("not a call in progress: %v", req.ID)
+	}
+
+	elapsed := time.Since(handling.start)
+	var raw *json.RawMessage
+	if err == nil {
+		raw, err = marshalToRaw(result)
+	}
+	response := &Response{
+		Result: raw,
+		ID:     req.ID,
+	}
+	if err != nil {
+		if callErr, ok := err.(*Error); ok {
+			response.Error = callErr
+		} else {
+			response.Error = NewErrorf(0, "%s", err)
+		}
+	}
+	data, err := json.Marshal(response)
+	if err != nil {
+		return err
+	}
+	c.log(Send, response.ID, elapsed, req.Method, response.Result, response.Error)
+	if err = c.stream.Write(ctx, data); err != nil {
+		// TODO(iancottrell): if a stream write fails, we really need to shut down
+		// the whole stream
+		return err
+	}
+	return nil
+}
+
+type handling struct {
+	request *Request
+	cancel  context.CancelFunc
+	start   time.Time
+}
+
 // combined has all the fields of both Request and Response.
 // We can decode this and then work out which it is.
 type combined struct {
@@ -230,7 +285,6 @@ type combined struct {
 // It must be called exactly once for each Conn.
 // It returns only when the reader is closed or there is an error in the stream.
 func (c *Conn) run(ctx context.Context) error {
-	ctx, cancelRun := context.WithCancel(ctx)
 	for {
 		// get the data for a message
 		data, err := c.stream.Read(ctx)
@@ -258,57 +312,19 @@ func (c *Conn) run(ctx context.Context) error {
 			if request.IsNotify() {
 				c.log(Receive, request.ID, -1, request.Method, request.Params, nil)
 				// we have a Notify, forward to the handler in a go routine
-				go func() {
+				c.handle(ctx, c, request)
-					if _, err := c.handle(ctx, c, request); err != nil {
-						// notify produced an error, we can't forward it to the other side
-						// because there is no id, so we just log it
-						c.log(Receive, nil, -1, request.Method, nil, err)
-					}
-				}()
 			} else {
 				// we have a Call, forward to the handler in another go routine
 				reqCtx, cancelReq := context.WithCancel(ctx)
 				c.handlingMu.Lock()
-				c.handling[*request.ID] = cancelReq
+				c.handling[*request.ID] = handling{
+					request: request,
+					cancel:  cancelReq,
+					start:   time.Now(),
+				}
 				c.handlingMu.Unlock()
-				go func() {
+				c.log(Receive, request.ID, -1, request.Method, request.Params, nil)
-					defer func() {
+				c.handle(reqCtx, c, request)
-						// clean up the cancel handler on the way out
-						c.handlingMu.Lock()
-						delete(c.handling, *request.ID)
-						c.handlingMu.Unlock()
-						cancelReq()
-					}()
-					c.log(Receive, request.ID, -1, request.Method, request.Params, nil)
-					before := time.Now()
-					resp, callErr := c.handle(reqCtx, c, request)
-					elapsed := time.Since(before)
-					var result *json.RawMessage
-					if result, err = marshalToRaw(resp); err != nil {
-						callErr = &Error{Message: err.Error()}
-					}
-					response := &Response{
-						Result: result,
-						Error:  callErr,
-						ID:     request.ID,
-					}
-					data, err := json.Marshal(response)
-					if err != nil {
-						// failure to marshal leaves the call without a response
-						// possibly we could attempt to respond with a different message
-						// but we can probably rely on timeouts instead
-						c.log(Send, request.ID, elapsed, request.Method, nil, NewErrorf(0, "%s", err))
-						return
-					}
-					c.log(Send, response.ID, elapsed, request.Method, response.Result, response.Error)
-					if err = c.stream.Write(ctx, data); err != nil {
-						// if a stream write fails, we really need to shut down the whole
-						// stream and return from the run
-						c.log(Send, request.ID, elapsed, request.Method, nil, NewErrorf(0, "%s", err))
-						cancelRun()
-						return
-					}
-				}()
 			}
 		case msg.ID != nil:
 			// we have a response, get the pending entry from the map

internal/jsonrpc2/jsonrpc2_test.go

@@ -102,7 +102,7 @@ func prepare(ctx context.Context, t *testing.T, withHeaders bool) (*testHandler,
 		} else {
 			h.stream = jsonrpc2.NewStream(h.reader, h.writer)
 		}
-		args := []interface{}{handle}
+		args := []interface{}{jsonrpc2.Handler(handle)}
 		if *logRPC {
 			args = append(args, jsonrpc2.Log)
 		}
@@ -128,32 +128,36 @@ type testHandler struct {
 	*jsonrpc2.Conn
 }
 
-func handle(ctx context.Context, c *jsonrpc2.Conn, r *jsonrpc2.Request) (interface{}, *jsonrpc2.Error) {
+func handle(ctx context.Context, c *jsonrpc2.Conn, r *jsonrpc2.Request) {
 	switch r.Method {
 	case "no_args":
 		if r.Params != nil {
-			return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeInvalidParams, "Expected no params")
+			c.Reply(ctx, r, nil, jsonrpc2.NewErrorf(jsonrpc2.CodeInvalidParams, "Expected no params"))
+			return
 		}
-		return true, nil
+		c.Reply(ctx, r, true, nil)
 	case "one_string":
 		var v string
 		if err := json.Unmarshal(*r.Params, &v); err != nil {
-			return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err.Error())
+			c.Reply(ctx, r, nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err.Error()))
+			return
 		}
-		return "got:" + v, nil
+		c.Reply(ctx, r, "got:"+v, nil)
 	case "one_number":
 		var v int
 		if err := json.Unmarshal(*r.Params, &v); err != nil {
-			return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err.Error())
+			c.Reply(ctx, r, nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err.Error()))
+			return
 		}
-		return fmt.Sprintf("got:%d", v), nil
+		c.Reply(ctx, r, fmt.Sprintf("got:%d", v), nil)
 	case "join":
 		var v []string
 		if err := json.Unmarshal(*r.Params, &v); err != nil {
-			return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err.Error())
+			c.Reply(ctx, r, nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err.Error()))
+			return
 		}
-		return path.Join(v...), nil
+		c.Reply(ctx, r, path.Join(v...), nil)
 	default:
-		return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeMethodNotFound, "method %q not found", r.Method)
+		c.Reply(ctx, r, nil, jsonrpc2.NewErrorf(jsonrpc2.CodeMethodNotFound, "method %q not found", r.Method))
 	}
 }

internal/lsp/protocol/client.go

@@ -25,121 +25,103 @@ type Client interface {
 }
 
 func clientHandler(client Client) jsonrpc2.Handler {
-	return func(ctx context.Context, conn *jsonrpc2.Conn, r *jsonrpc2.Request) (interface{}, *jsonrpc2.Error) {
+	return func(ctx context.Context, conn *jsonrpc2.Conn, r *jsonrpc2.Request) {
 		switch r.Method {
 		case "$/cancelRequest":
 			var params CancelParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			conn.Cancel(params.ID)
-			return nil, nil
 
 		case "window/showMessage":
 			var params ShowMessageParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
-			if err := client.ShowMessage(ctx, &params); err != nil {
+			unhandledError(client.ShowMessage(ctx, &params))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		case "window/showMessageRequest":
 			var params ShowMessageRequestParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := client.ShowMessageRequest(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "window/logMessage":
 			var params LogMessageParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
-			if err := client.LogMessage(ctx, &params); err != nil {
+			unhandledError(client.LogMessage(ctx, &params))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		case "telemetry/event":
 			var params interface{}
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
-			if err := client.Telemetry(ctx, &params); err != nil {
+			unhandledError(client.Telemetry(ctx, &params))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		case "client/registerCapability":
 			var params RegistrationParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
-			if err := client.RegisterCapability(ctx, &params); err != nil {
+			unhandledError(client.RegisterCapability(ctx, &params))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		case "client/unregisterCapability":
 			var params UnregistrationParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
-			if err := client.UnregisterCapability(ctx, &params); err != nil {
+			unhandledError(client.UnregisterCapability(ctx, &params))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		case "workspace/workspaceFolders":
 			if r.Params != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeInvalidParams, "Expected no params")
+				conn.Reply(ctx, r, nil, jsonrpc2.NewErrorf(jsonrpc2.CodeInvalidParams, "Expected no params"))
+				return
 			}
 			resp, err := client.WorkspaceFolders(ctx)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "workspace/configuration":
 			var params ConfigurationParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := client.Configuration(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "workspace/applyEdit":
 			var params ApplyWorkspaceEditParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := client.ApplyEdit(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/publishDiagnostics":
 			var params PublishDiagnosticsParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
-			if err := client.PublishDiagnostics(ctx, &params); err != nil {
+			unhandledError(client.PublishDiagnostics(ctx, &params))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		default:
-			return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeMethodNotFound, "method %q not found", r.Method)
+			if r.IsNotify() {
+				conn.Reply(ctx, r, nil, jsonrpc2.NewErrorf(jsonrpc2.CodeMethodNotFound, "method %q not found", r.Method))
+			}
 		}
 	}
 }

internal/lsp/protocol/protocol.go

@@ -6,6 +6,7 @@ package protocol
 
 import (
 	"context"
+	"log"
 
 	"golang.org/x/tools/internal/jsonrpc2"
 )
@@ -15,20 +16,34 @@ func canceller(ctx context.Context, conn *jsonrpc2.Conn, req *jsonrpc2.Request)
 }
 
 func RunClient(ctx context.Context, stream jsonrpc2.Stream, client Client, opts ...interface{}) (*jsonrpc2.Conn, Server) {
-	opts = append([]interface{}{clientHandler(client), canceller}, opts...)
+	opts = append([]interface{}{clientHandler(client), jsonrpc2.Canceler(canceller)}, opts...)
 	conn := jsonrpc2.NewConn(ctx, stream, opts...)
 	return conn, &serverDispatcher{Conn: conn}
 }
 
 func RunServer(ctx context.Context, stream jsonrpc2.Stream, server Server, opts ...interface{}) (*jsonrpc2.Conn, Client) {
-	opts = append([]interface{}{serverHandler(server), canceller}, opts...)
+	opts = append([]interface{}{serverHandler(server), jsonrpc2.Canceler(canceller)}, opts...)
 	conn := jsonrpc2.NewConn(ctx, stream, opts...)
 	return conn, &clientDispatcher{Conn: conn}
 }
 
-func toJSONError(err error) *jsonrpc2.Error {
+func sendParseError(ctx context.Context, conn *jsonrpc2.Conn, req *jsonrpc2.Request, err error) {
-	if jsonError, ok := err.(*jsonrpc2.Error); ok {
+	if _, ok := err.(*jsonrpc2.Error); !ok {
-		return jsonError
+		err = jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
 	}
-	return jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+	unhandledError(conn.Reply(ctx, req, nil, err))
+}
+
+// unhandledError is used in places where an error may occur that cannot be handled.
+// This occurs in things like rpc handlers that are a notify, where we cannot
+// reply to the caller, or in a call when we are actually attempting to reply.
+// In these cases, there is nothing we can do with the error except log it, so
+// we do that in this function, and the presence of this function acts as a
+// useful reminder of why we are effectively dropping the error and also a
+// good place to hook in when debugging those kinds of errors.
+func unhandledError(err error) {
+	if err == nil {
+		return
+	}
+	log.Printf("%v", err)
 }

internal/lsp/protocol/server.go

@@ -52,411 +52,339 @@ type Server interface {
 }
 
 func serverHandler(server Server) jsonrpc2.Handler {
-	return func(ctx context.Context, conn *jsonrpc2.Conn, r *jsonrpc2.Request) (interface{}, *jsonrpc2.Error) {
+	return func(ctx context.Context, conn *jsonrpc2.Conn, r *jsonrpc2.Request) {
 		switch r.Method {
 		case "initialize":
 			var params InitializeParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.Initialize(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "initialized":
 			var params InitializedParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
-			if err := server.Initialized(ctx, &params); err != nil {
+			unhandledError(server.Initialized(ctx, &params))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		case "shutdown":
 			if r.Params != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeInvalidParams, "Expected no params")
+				conn.Reply(ctx, r, nil, jsonrpc2.NewErrorf(jsonrpc2.CodeInvalidParams, "Expected no params"))
+				return
 			}
-			if err := server.Shutdown(ctx); err != nil {
+			unhandledError(server.Shutdown(ctx))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		case "exit":
 			if r.Params != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeInvalidParams, "Expected no params")
+				conn.Reply(ctx, r, nil, jsonrpc2.NewErrorf(jsonrpc2.CodeInvalidParams, "Expected no params"))
+				return
 			}
-			if err := server.Exit(ctx); err != nil {
+			unhandledError(server.Exit(ctx))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		case "$/cancelRequest":
 			var params CancelParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			conn.Cancel(params.ID)
-			return nil, nil
 
 		case "workspace/didChangeWorkspaceFolders":
 			var params DidChangeWorkspaceFoldersParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
-			if err := server.DidChangeWorkspaceFolders(ctx, &params); err != nil {
+			unhandledError(server.DidChangeWorkspaceFolders(ctx, &params))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		case "workspace/didChangeConfiguration":
 			var params DidChangeConfigurationParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
-			if err := server.DidChangeConfiguration(ctx, &params); err != nil {
+			unhandledError(server.DidChangeConfiguration(ctx, &params))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		case "workspace/didChangeWatchedFiles":
 			var params DidChangeWatchedFilesParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
-			if err := server.DidChangeWatchedFiles(ctx, &params); err != nil {
+			unhandledError(server.DidChangeWatchedFiles(ctx, &params))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		case "workspace/symbol":
 			var params WorkspaceSymbolParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.Symbols(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "workspace/executeCommand":
 			var params ExecuteCommandParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.ExecuteCommand(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/didOpen":
 			var params DidOpenTextDocumentParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
-			if err := server.DidOpen(ctx, &params); err != nil {
+			unhandledError(server.DidOpen(ctx, &params))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		case "textDocument/didChange":
 			var params DidChangeTextDocumentParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
-			if err := server.DidChange(ctx, &params); err != nil {
+			unhandledError(server.DidChange(ctx, &params))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		case "textDocument/willSave":
 			var params WillSaveTextDocumentParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
-			if err := server.WillSave(ctx, &params); err != nil {
+			unhandledError(server.WillSave(ctx, &params))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		case "textDocument/willSaveWaitUntil":
 			var params WillSaveTextDocumentParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.WillSaveWaitUntil(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/didSave":
 			var params DidSaveTextDocumentParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
-			if err := server.DidSave(ctx, &params); err != nil {
+			unhandledError(server.DidSave(ctx, &params))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		case "textDocument/didClose":
 			var params DidCloseTextDocumentParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
-			if err := server.DidClose(ctx, &params); err != nil {
+			unhandledError(server.DidClose(ctx, &params))
-				return nil, toJSONError(err)
-			}
-			return nil, nil
 
 		case "textDocument/completion":
 			var params CompletionParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.Completion(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "completionItem/resolve":
 			var params CompletionItem
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.CompletionResolve(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/hover":
 			var params TextDocumentPositionParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.Hover(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/signatureHelp":
 			var params TextDocumentPositionParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.SignatureHelp(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/definition":
 			var params TextDocumentPositionParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.Definition(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/typeDefinition":
 			var params TextDocumentPositionParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.TypeDefinition(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/implementation":
 			var params TextDocumentPositionParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.Implementation(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/references":
 			var params ReferenceParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.References(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/documentHighlight":
 			var params TextDocumentPositionParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.DocumentHighlight(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/documentSymbol":
 			var params DocumentSymbolParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.DocumentSymbol(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/codeAction":
 			var params CodeActionParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.CodeAction(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/codeLens":
 			var params CodeLensParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.CodeLens(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "codeLens/resolve":
 			var params CodeLens
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.CodeLensResolve(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/documentLink":
 			var params DocumentLinkParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.DocumentLink(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "documentLink/resolve":
 			var params DocumentLink
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.DocumentLinkResolve(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/documentColor":
 			var params DocumentColorParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.DocumentColor(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/colorPresentation":
 			var params ColorPresentationParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.ColorPresentation(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/formatting":
 			var params DocumentFormattingParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.Formatting(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/rangeFormatting":
 			var params DocumentRangeFormattingParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.RangeFormatting(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/onTypeFormatting":
 			var params DocumentOnTypeFormattingParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.OnTypeFormatting(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/rename":
 			var params RenameParams
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.Rename(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 
 		case "textDocument/foldingRanges":
 			var params FoldingRangeRequestParam
 			if err := json.Unmarshal(*r.Params, &params); err != nil {
-				return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeParseError, "%v", err)
+				sendParseError(ctx, conn, r, err)
+				return
 			}
 			resp, err := server.FoldingRanges(ctx, &params)
-			if err != nil {
+			unhandledError(conn.Reply(ctx, r, resp, err))
-				return nil, toJSONError(err)
-			}
-			return resp, nil
 		default:
-			return nil, jsonrpc2.NewErrorf(jsonrpc2.CodeMethodNotFound, "method %q not found", r.Method)
+			if r.IsNotify() {
+				conn.Reply(ctx, r, nil, jsonrpc2.NewErrorf(jsonrpc2.CodeMethodNotFound, "method %q not found", r.Method))
+			}
 		}
 	}
 }