Fix 271

answer.go

@@ -2,6 +2,7 @@ package capnp
 
 import (
  "context"
+ "fmt"
  "strconv"
  "sync"
 
@@ -128,9 +129,7 @@ func (p *Promise) resolution() resolution {
 func (p *Promise) Fulfill(result Ptr) {
  defer p.mu.Unlock()
  p.mu.Lock()
- if !p.isUnresolved() {
- panic("Promise.Fulfill called after Fulfill or Reject")
- }
+ p.requireUnresolved("Fulfill")
  p.resolve(result, nil)
 }
 
@@ -145,9 +144,7 @@ func (p *Promise) Reject(e error) {
  }
  defer p.mu.Unlock()
  p.mu.Lock()
- if !p.isUnresolved() {
- panic("Promise.Reject called after Fulfill or Reject")
- }
+ p.requireUnresolved("Reject")
  p.resolve(Ptr{}, e)
 }
 
@@ -163,6 +160,28 @@ func (p *Promise) Resolve(r Ptr, e error) {
  }
 }
 
+// requireUnresolved is a helper method for checking for duplicate
+// calls to Fulfill() or Reject(); panics if the promise is not in
+// the unresolved state.
+//
+// The callerMethod argument should be the name of the method which
+// is invoking requireUnresolved. The panic message will report this
+// value as well as the method that originally resolved the promise,
+// and which method (Fulfill or Reject) was used to resolve it.
+func (p *Promise) requireUnresolved(callerMethod string) {
+ if !p.isUnresolved() {
+ var prevMethod string
+ if p.err == nil {
+ prevMethod = "Fulfill"
+ } else {
+ prevMethod = fmt.Sprintf("Reject (error = %q)", p.err)
+ }
+
+ panic("Promise." + callerMethod +
+ " called after previous call to " + prevMethod)
+ }
+}
+
 // resolve moves p into the resolved state from unresolved. The caller
 // must be holding onto p.mu.
 func (p *Promise) resolve(r Ptr, e error) {

rpc/level0_test.go

@@ -15,6 +15,7 @@ import (
  "capnproto.org/go/capnp/v3/exc"
  "capnproto.org/go/capnp/v3/pogs"
  "capnproto.org/go/capnp/v3/rpc"
+ testcp "capnproto.org/go/capnp/v3/rpc/internal/testcapnp"
  "capnproto.org/go/capnp/v3/rpc/transport"
  "capnproto.org/go/capnp/v3/server"
  rpccp "capnproto.org/go/capnp/v3/std/capnp/rpc"
@@ -1768,6 +1769,58 @@ func TestSendCancel(t *testing.T) {
  }
 }
 
+func TestHandleReturn_regression(t *testing.T) {
+ t.Parallel()
+
+ p1, p2 := transport.NewPipe(1)
+
+ srv := testcp.PingPong_ServerToClient(pingPongServer{})
+ conn1 := rpc.NewConn(rpc.NewTransport(p2), &rpc.Options{
+ BootstrapClient: capnp.Client(srv),
+ })
+ defer conn1.Close()
+
+ conn2 := rpc.NewConn(rpc.NewTransport(p1), nil)
+ defer conn2.Close()
+
+ t.Run("MethodCallWithExpiredContext", func(t *testing.T) {
+ pp := testcp.PingPong(conn2.Bootstrap(context.Background()))
+ defer pp.Release()
+
+ // create an EXPIRED context
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+
+ f, release := pp.EchoNum(ctx, func(ps testcp.PingPong_echoNum_Params) error {
+ ps.SetN(42)
+ return nil
+ })
+ defer release()
+
+ _, err := f.Struct()
+ assert.ErrorIs(t, err, ctx.Err())
+ })
+
+ t.Run("BootstrapWithExpiredContext", func(t *testing.T) {
+ // create an EXPIRED context
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+
+ // NOTE: bootstrap with expired context
+ pp := testcp.PingPong(conn2.Bootstrap(ctx))
+ defer pp.Release()
+
+ f, release := pp.EchoNum(ctx, func(ps testcp.PingPong_echoNum_Params) error {
+ ps.SetN(42)
+ return nil
+ })
+ defer release()
+
+ _, err := f.Struct()
+ assert.ErrorIs(t, err, ctx.Err())
+ })
+}
+
 // finishTest drains both sides of a pipe and reports any errors to t.
 func finishTest(t errorfer, conn *rpc.Conn, p2 rpc.Transport) {
  ctx, cancel := context.WithCancel(context.Background())

rpc/question.go

@@ -157,9 +157,11 @@ func (q *question) PipelineSend(ctx context.Context, transform []capnp.PipelineO
  return q.c.newPipelineCallMessage(m, q.id, transform, q2.id, s)
  }, func(err error) {
  if err != nil {
- q.c.questions[q2.id] = nil
- q.c.questionID.remove(uint32(q2.id))
- q.p.Reject(rpcerr.Failedf("send message: %w", err))
+ syncutil.With(&q.c.mu, func() {
+ q.c.questions[q2.id] = nil
+ q.c.questionID.remove(uint32(q2.id))
+ })
+ q2.p.Reject(rpcerr.Failedf("send message: %w", err))
  return
  }