Repository: creachadair/jrpc2
Pull request #13 (2020-03-17 01:04:53)

server: Allow caller to distinguish shutdown from close
Addresses #11. The Wait method filters out normal channel close errors, along
with explicit calls to Stop.  This is usually what you want, but means the
caller can't distinguish explicit shutdown from channel closes.

Add a WaitStatus method that includes more detail, and re-implement Wait in
terms of it.

radeksimko: I assume you meant to reference #11 rather than #12 

--- 

I like the idea and it does work for simple (and common) use case with single pair of I/O reader and writer, but I think the `server.Loop` may need to account for that too

https://github.com/creachadair/jrpc2/blob/7a9729d0493527e216051e844c9144eb50a22ac0/server/loop.go#L53-L55

which makes me think the error checking there is probably already flawed - since `Wait()` will never pass through `io.EOF`?

I generally understand your sentiment about overusing hooks, but I'm thinking they may be unavoidable here in the `Loop` server? I'm open to any alternatives though.

For context I use the Loop server in the second of two modes in which our Language Server can operate - stdin/stdout and TCP exposed over a port. I believe that in case of TCP each connection should have its own lifecycle which is separate from the "server abstraction" itself dispensing `Server` instances for each connection. For that reason I believe that there should be a way of performing connection-level cleanup, rather than allowing cleanup after _all_ connections are closed.

So how do you feel about adding

```go
ServerDone(jrpc2.ServerStatus)
```
to `LoopOptions`? Perhaps knowing what connection is being closed is also useful - so the ideal interface may be
```go
ServerDone(net.Conn, jrpc2.ServerStatus)
```
but that would expose the whole connection unnecessarily, so maybe just this?
```go
ServerDone(localAddr net.Addr, status jrpc2.ServerStatus)
```
To be fair though the only use case I have so far for identifying the connection is troubleshooting - I would likely end up just adding a log message in the form of `Connection from <net.Addr> closed`. I don't know if that's strong enough reason to justify the second argument. 🤷‍♂ 
radeksimko: I just realized I have another use case for the last mentioned interface

```go
ServerDone(localAddr net.Addr, status jrpc2.ServerStatus)
```

In LSP there is need to maintain state of the server (where methods like `initialize`, `initialized`, or `shutdown` can mutate it) to ensure compliance with the spec - e.g. that a server is only initialized once per workspace.

This is easy to achieve with a single instance as clients just launch new process and the state is maintained per process. It gets more difficult if there's a single process listening for many TCP connections - a strong need for identifying individual connection and their state arises there.

Relatedly I see use case for `ServerAccept`, but that's for a separate PR I suppose and not 
creachadair: > I assume you meant to reference #11 rather than #12

Yes, sorry, fixed.

> https://github.com/creachadair/jrpc2/blob/7a9729d0493527e216051e844c9144eb50a22ac0/server/loop.go#L53-L55
>
> which makes me think the error checking there is probably already flawed - since `Wait()` will never pass through `io.EOF`?

It won't, as you say. I think this is an artifact of an earlier version of the code, where I didn't filter on those errors. That means it's wrong twice: Once because it _only_ checks for `io.EOF` and not channel closed, and once because it _does_ check for `io.EOF`. 🙂

One of the main reasons `Wait` filters those conditions now is that I found ~every application had to add those checks explicitly. Apparently I didn't clean them all up.

> [I]n case of TCP each connection should have its own lifecycle which is separate from the "server abstraction" itself dispensing `Server` instances for each connection.

I'm not sure I understand the motivation for this: Are you saying you want to handle multiplexing requests from multiple clients into a single `Server` instance? That is possible, but tedious—and I don't think it should be necessary. So first let me ask, is that what you're describing?

You mentioned shared state for LSP, which makes me wonder if that's part of the issue. If so, what I recommend is hooking the state management to a service type, not the server itself. For example (in outline):

```go
type Service struct {
  once sync.Once
  shutdown struct {
    sync.Mutex
    inProgress bool
  }
  // other fields
}

func (s *Service) init() { 
  // ... various one-time initializations ...
}

func (s *Service) Initialize(ctx context.Context, req *InitReq) (*InitRsp, error) {
  s.once.Do(s.init) // ensures this only happens once
  // ...
  return rsp, nil
}

func (s *Service) Shutdown(ctx context.Context, req *ShutdownReq) error {
  s.shutdown.Lock()
  defer s.shutdown.Unlock()
  s.shutdown.inProgress = true
  // whatever else needs to happen to trigger shutdown
  return nil
}

func (s *Service) Other(ctx context.Context) error {
  if s.isShuttingDown() {
    return code.InvalidRequest.Err() // 
  }
  // ... etc.
}

func (s *Service) isShuttingDown() bool {
  s.shutdown.Lock(); defer s.shutdown.Unlock(); return s.shutdown.inProgress
}

```

The methods can manage state in the usual ways—locks to gate concurrent access, channels to pass work to a maintenance goroutine, atomics for counters, etc. That is likely to be less fragile than trying to carefully coordinate with server lifecycle events.

> For that reason I believe that there should be a way of performing connection-level cleanup, rather than allowing cleanup after _all_ connections are closed.

I'm having trouble understanding what additional per-connection cleanup is needed, can you expand on that? Each `Accept` call returns a new, separate `net.Conn` value, independent of its originating socket. The channel and the server (are intended to) share fate: Closing the channel ends the server (in [`(*Server).read`](https://github.com/creachadair/jrpc2/blob/7a9729d0493527e216051e844c9144eb50a22ac0/server.go#L425)), and stopping the server closes the channel (in [`(*Server).stop`](https://github.com/creachadair/jrpc2/blob/7a9729d0493527e216051e844c9144eb50a22ac0/server.go#L380)). Is there some other signal the channel needs?

> This is easy to achieve with a single instance as clients just launch new process and the state is maintained per process. It gets more difficult if there's a single process listening for many TCP connections - a strong need for identifying individual connection and its state arises there.

I believe managing per-process state is orthogonal: Even with many connections, the  implementation can manage concurrent access to its global state independently from the server plumbing.

I'm not dead-set against adding a callback, and I'm more comfortable adding one to `LoopOptions` than to `ServerOptions`. Still, I'd like to understand the use cases a bit better first.
creachadair: I was reflecting a bit more on these comments:

> In LSP there is need to maintain state of the server (where methods like `initialize`, `initialized`, or `shutdown` can mutate it) to ensure compliance with the spec - e.g. that a server is only initialized once per workspace.
> 
> This is easy to achieve with a single instance as clients just launch new process and the state is maintained per process. It gets more difficult if there's a single process listening for many TCP connections - a strong need for identifying individual connection and its state arises there.

I think I was distracted by the focus on _server_ state, but now I wonder: Is the issue not the global state of the server process, but per-connection state that is _not_ global to the process?
radeksimko: > Are you saying you want to handle multiplexing requests from multiple clients into a single Server instance? That is possible, but tedious—and I don't think it should be necessary. So first let me ask, is that what you're describing?

Not quite. I'll try to explain below.

> If so, what I recommend is hooking the state management to a service type

I have something very similar to your code actually. If I'm interpreting it correctly though both implementations only work in a scenario where 1 process = 1 client = 1 state, because the state is global, mutable only by that single client, therefore kinda unsuitable for `server.Loop` which is by design handling multiple connections/clients and therefore multiple states.

>  Is the issue not the global state of the server process, but per-connection state that is not global to the process?

Exactly 💯 

I think it's not documented in the spec (probably because the spec doesn't document TCP-based server at all), but expected behaviour (by some clients) when server runs in TCP mode. At that point each new connection represents a new workspace in an IDE, but is handled by the same process. Each workspace (and hence connection) has its own state (will be `initialize`d separately).

--- 

Apologies for the misunderstanding, I hope the above cleared it up a bit more.

Also I noticed I made an error in my previous code example (now fixed) - the connection is uniquely identified by its remote address (from server's perspective), rather than local - the local address is server's which is always the same.

Also we will be open-sourcing the language server in a couple of weeks, but I'd be also happy to get you invited to the private repo in the meantime and point you to the code, if that helps the discussion here.
creachadair: > > Is the issue not the global state of the server process, but per-connection state that is not global to the process?
> 
> Exactly 💯

Ok, I think that resolves my confusion.

> I think it's not documented in the spec (probably because the spec doesn't document TCP-based server at all), but expected behaviour (by some clients) when server runs in TCP mode. At that point each new connection represents a new workspace in an IDE, but is handled by the same process. Each workspace (and hence connection) has its own state (will be `initialize`d separately).

Yes, that makes sense.

For this case, I think it might make more sense _not_ to use `server.Loop`, but to write your own loop that handles this case. Instead of constructing a single "service" (the value that has the handlers on it) when the process starts, each _connection_ would get its own service value, and thus would "own" all the state specific to its invocation. 

If there is shared global state, it can be shared through a global variable.

The following isn't tested, but illustrates what I mean:

```go
type Service struct {
  // ...
}

func (s *Service) assign() (jrpc2.Assigner, error) {
  // called once when the service on this value is about to start
  // also bind names to methods
}
func (s *Service) finish(stat jrpc2.ServerStatus) {
  // called when the service on this value has exited
}

func (s *Service) Initialize(ctx context.Context, req *InitReq) (*InitRsp, error) { /* ... */ }

func Run(lst net.Listener, newService func() *Service, opts *RunOptions) {
   var wg sync.WaitGroup
   for {
      conn, err := lst.Accept()
      if err != nil {
         if channel.IsErrClosing(err) {
            err = nil
         } else {
            log("Error accepting new connection: %v", err)
         }
         wg.Wait()
         return err
      }
      ch := newChannel(conn, conn)
      wg.Add(1)
      go func() {
         defer wg.Done()
         svc := newService()
         ass, err := svc.assign()
         if err != nil {
            log("Service initialization failed: %v", err)
            return
         }
         srv := jrpc2.NewServer(ass, opts.ServerOptions).Start(ch)
         stat := srv.WaitStatus()
         svc.finish(stat)
         if stat.Err != nil {
            log("Server exit: %v", stat.Err)
         }
      }()
   }
}
```

This is roughly the same code that's in `server.Loop` now, but with some of the plumbing skipped, and a slightly different interface. The main difference is that each server invocation gets a _fresh_ state from a constructor. The constructor can plug in references to any shared global state that may be relevant, or just create a new empty value.

The main thing the server doesn't _currently_ have to support this (unless we merge this PR or something like it) is a way to distinguish how the service exited.

Granted, this would be an additional package you'd have to maintain, and already I can see a couple improvements to the interface that I'd make if I were going to build this for real, but I think it's not too big a maintenance headache and allows you to have multiple distinct servers without too much pain.

> Also we will be open-sourcing the language server in a couple of weeks, but I'd be also happy to get you invited to the private repo in the meantime and point you to the code, if that helps the discussion here.

If it turns out I'm still missing some essential point, I'm certainly willing to take a look. 🙂 
creachadair: > > For this case, I think it might make more sense not to use server.Loop, but to write your own loop that handles this somewhat-uncommon situation.
> 
> 👍 That is fair and I'm not against that idea at all.

It could even be made fairly generic:

```go
type Service interface {
   Assigner() (jrpc2.Assigner, error)
   Finish(jrpc2.ServerStatus)
}

func Run(lst net.Listener, newService func() Service, opts *RunOptions) error {
  // ...
}
```

…which would allow any existing single-use service type to be decorated with these methods to make it eligible for use here. Anyway, let me know if that causes other problems, and we can certainly discuss other options. Meanwhile I will go ahead and merge this, since I think it will be useful anyway.

> In that case this PR LGTM. Thank you for the discussion and time you spent on this.

No problem. I'm grateful for the feedback you've provided throughout, it's very helpful to have another brain thinking about these questions.

Patch:
diff --git a/server.go b/server.go
index 85d2783..162df24 100644
--- a/server.go
+++ b/server.go
@@ -358,22 +358,43 @@ func (s *Server) Stop() {
 	s.stop(errServerStopped)
 }
 
-// Wait blocks until the connection terminates and returns the resulting error.
-// After Wait returns, whether or not there was an error, it is safe to call
-// s.Start again to restart the server with a fresh channel.
-func (s *Server) Wait() error {
+// ServerStatus describes the status of a stopped server.
+type ServerStatus struct {
+	Err error // the error that caused the server to stop (nil on success)
+
+	stopped bool // whether Stop was called
+}
+
+// Success reports whether the server exited without error.
+func (s ServerStatus) Success() bool { return s.Err == nil }
+
+// Stopped reports whether the server exited due to Stop being called.
+func (s ServerStatus) Stopped() bool { return s.Err == nil && s.stopped }
+
+// Closed reports whether the server exited due to a channel close.
+func (s ServerStatus) Closed() bool { return s.Err == nil && !s.stopped }
+
+// WaitStatus blocks until the server terminates, and returns the resulting
+// status. After WaitStatus returns, whether or not there was an error, it is
+// safe to call s.Start again to restart the server with a fresh channel.
+func (s *Server) WaitStatus() ServerStatus {
 	s.wg.Wait()
 	// Sanity check.
 	if s.inq.Len() != 0 {
 		panic("s.inq is not empty at shutdown")
 	}
+	exitErr := s.err
 	// Don't remark on a closed channel or EOF as a noteworthy failure.
 	if s.err == io.EOF || channel.IsErrClosing(s.err) || s.err == errServerStopped {
-		return nil
+		exitErr = nil
 	}
-	return s.err
+	return ServerStatus{Err: exitErr, stopped: s.err == errServerStopped}
 }
 
+// Wait blocks until the server terminates and returns the resulting error.
+// It is equivalent to s.WaitStatus().Err.
+func (s *Server) Wait() error { return s.WaitStatus().Err }
+
 // stop shuts down the connection and records err as its final state.  The
 // caller must hold s.mu. If multiple callers invoke stop, only the first will
 // successfully record its error status.


Test patch:
diff --git a/jrpc2_test.go b/jrpc2_test.go
index 00cc484..87981ab 100644
--- a/jrpc2_test.go
+++ b/jrpc2_test.go
@@ -867,3 +867,49 @@ type assignFunc func(context.Context, string) jrpc2.Handler
 
 func (a assignFunc) Assign(ctx context.Context, m string) jrpc2.Handler { return a(ctx, m) }
 func (assignFunc) Names() []string                                      { return nil }
+
+func TestWaitStatus(t *testing.T) {
+	check := func(t *testing.T, stat jrpc2.ServerStatus, closed, stopped bool, wantErr error) {
+		t.Helper()
+		t.Logf("Server status: %+v", stat)
+		if got, want := stat.Success(), wantErr == nil; got != want {
+			t.Errorf("Status success: got %v, want %v", got, want)
+		}
+		if got := stat.Closed(); got != closed {
+			t.Errorf("Status closed: got %v, want %v", got, closed)
+		}
+		if got := stat.Stopped(); got != stopped {
+			t.Errorf("Status stopped: got %v, want %v", got, stopped)
+		}
+		if stat.Err != wantErr {
+			t.Errorf("Status error: got %v, want %v", stat.Err, wantErr)
+		}
+	}
+	t.Run("ChannelClosed", func(t *testing.T) {
+		loc := server.NewLocal(handler.Map{"OK": testOK}, nil)
+		loc.Client.Close()
+		check(t, loc.Server.WaitStatus(), true, false, nil)
+	})
+
+	t.Run("ServerStopped", func(t *testing.T) {
+		loc := server.NewLocal(handler.Map{"OK": testOK}, nil)
+		loc.Server.Stop()
+		check(t, loc.Server.WaitStatus(), false, true, nil)
+	})
+
+	t.Run("ChannelFailed", func(t *testing.T) {
+		wantErr := errors.New("failed")
+		ch := buggyChannel{data: "bogus", err: wantErr}
+		srv := jrpc2.NewServer(handler.Map{"OK": testOK}, nil).Start(ch)
+		check(t, srv.WaitStatus(), false, false, wantErr)
+	})
+}
+
+type buggyChannel struct {
+	data string
+	err  error
+}
+
+func (buggyChannel) Send([]byte) error       { panic("should not be called") }
+func (b buggyChannel) Recv() ([]byte, error) { return []byte(b.data), b.err }
+func (buggyChannel) Close() error            { return nil }