feat(server): return actual error when context done (#581)

### Motivation

Return the actual errors if the context is done. 

### Modification

- Return the actual error to client.
- normal close stream if client closed stream.

server/leader_controller.go

@@ -35,6 +35,8 @@ import (
 	"github.com/streamnative/oxia/server/wal"
 )
 
+var ErrLeaderClosed = errors.New("the leader has been closed")
+
 type GetResult struct {
 	Response *proto.GetResponse
 	Err      error
@@ -874,9 +876,9 @@ func (lc *leaderController) WriteStream(stream proto.OxiaClient_WriteStreamServe
 	case err := <-closeStreamCh:
 		return err
 	case <-stream.Context().Done():
-		return context.Canceled
+		return stream.Context().Err()
 	case <-lc.ctx.Done():
-		return context.Canceled
+		return errors.Wrap(ErrLeaderClosed, lc.ctx.Err().Error())
 	}
 }
 

server/public_rpc_server.go

@@ -141,14 +141,17 @@ func (s *publicRpcServer) WriteStream(stream proto.OxiaClient_WriteStreamServer)
 		return err
 	}
 
-	err = lc.WriteStream(stream)
-	if err != nil &&
-		!errors.Is(err, io.EOF) &&
-		!errors.Is(err, context.Canceled) {
-		log.Warn(
-			"Write stream failed",
-			slog.Any("error", err),
-		)
+	if err = lc.WriteStream(stream); err != nil {
+		if errors.Is(err, io.EOF) || errors.Is(err, context.Canceled) {
+			log.Info("Write stream has been closed")
+			return nil
+		}
+		if errors.Is(err, ErrLeaderClosed) {
+			log.Info("Write stream has been closed by leader closed")
+			return err
+		}
+		log.Warn("Write stream has been closed by error", slog.Any("error", err))
+		return err
 	}
 	return err
 }