Bugfix/unbounded send queue

.github/workflows/go.yml

@@ -19,4 +19,5 @@ jobs:
         uses: actions/setup-go@v1
         with:
           go-version: ${{ matrix.go }}
-      - run: go test -v -race ./...
+      - run: go test -v -race -short ./...
+      - run: go test -v ./...

answer.go

@@ -5,7 +5,7 @@ import (
 	"strconv"
 	"sync"
 
-	"capnproto.org/go/capnp/v3/internal/errors"
+	"capnproto.org/go/capnp/v3/exc"
 	"capnproto.org/go/capnp/v3/internal/syncutil"
 )
 
@@ -766,11 +766,11 @@ type resolution struct {
 // ptr obtains a Ptr by applying a transform.
 func (r resolution) ptr(transform []PipelineOp) (Ptr, error) {
 	if r.err != nil {
-		return Ptr{}, errors.Annotate("", r.method.String(), r.err)
+		return Ptr{}, exc.Annotate("", r.method.String(), r.err)
 	}
 	p, err := Transform(r.result, transform)
 	if err != nil {
-		return Ptr{}, errors.Annotate("", r.method.String(), err)
+		return Ptr{}, exc.Annotate("", r.method.String(), err)
 	}
 	return p, nil
 }
@@ -789,7 +789,7 @@ func (r resolution) client(transform []PipelineOp) *Client {
 	}
 	iface := p.Interface()
 	if p.IsValid() && !iface.IsValid() {
-		return ErrorClient(newError("not a capability"))
+		return ErrorClient(errorf("not a capability"))
 	}
 	return iface.Client()
 }

canonical.go

@@ -17,7 +17,7 @@ func Canonicalize(s Struct) ([]byte, error) {
 		return nil, errorf("canonicalize: %v", err)
 	}
 	if err := fillCanonicalStruct(root, s); err != nil {
-		return nil, annotate(err).errorf("canonicalize")
+		return nil, annotatef(err, "canonicalize")
 	}
 	return seg.Data(), nil
 }
@@ -55,14 +55,14 @@ func fillCanonicalStruct(dst, s Struct) error {
 	for i := uint16(0); i < dst.size.PointerCount; i++ {
 		p, err := s.Ptr(i)
 		if err != nil {
-			return annotate(err).errorf("struct pointer %d", i)
+			return annotatef(err, "struct pointer %d", i)
 		}
 		cp, err := canonicalPtr(dst.seg, p)
 		if err != nil {
-			return annotate(err).errorf("struct pointer %d", i)
+			return annotatef(err, "struct pointer %d", i)
 		}
 		if err := dst.SetPtr(i, cp); err != nil {
-			return annotate(err).errorf("struct pointer %d", i)
+			return annotatef(err, "struct pointer %d", i)
 		}
 	}
 	return nil
@@ -124,7 +124,7 @@ func canonicalList(dst *Segment, l List) (List, error) {
 			}
 			cp, err := canonicalPtr(dst, p)
 			if err != nil {
-				return List{}, annotate(err).errorf("list element %d", i)
+				return List{}, annotatef(err, "list element %d", i)
 			}
 			if err := cl.Set(i, cp); err != nil {
 				return List{}, errorf("list element %d: %v", i, err)
@@ -150,7 +150,7 @@ func canonicalList(dst *Segment, l List) (List, error) {
 	}
 	for i := 0; i < cl.Len(); i++ {
 		if err := fillCanonicalStruct(cl.Struct(i), l.Struct(i)); err != nil {
-			return List{}, annotate(err).errorf("list element %d", i)
+			return List{}, annotatef(err, "list element %d", i)
 		}
 	}
 	return cl, nil

capability.go

@@ -285,10 +285,10 @@ func (c *Client) SendCall(ctx context.Context, s Send) (*Answer, ReleaseFunc) {
 	h, _, released, finish := c.startCall()
 	defer finish()
 	if released {
-		return ErrorAnswer(s.Method, newError("call on released client")), func() {}
+		return ErrorAnswer(s.Method, errorf("call on released client")), func() {}
 	}
 	if h == nil {
-		return ErrorAnswer(s.Method, newError("call on null client")), func() {}
+		return ErrorAnswer(s.Method, errorf("call on null client")), func() {}
 	}
 
 	limiter := c.GetFlowLimiter()
@@ -345,11 +345,11 @@ func (c *Client) RecvCall(ctx context.Context, r Recv) PipelineCaller {
 	h, _, released, finish := c.startCall()
 	defer finish()
 	if released {
-		r.Reject(newError("call on released client"))
+		r.Reject(errorf("call on released client"))
 		return nil
 	}
 	if h == nil {
-		r.Reject(newError("call on null client"))
+		r.Reject(errorf("call on null client"))
 		return nil
 	}
 	return h.Recv(ctx, r)
@@ -384,11 +384,13 @@ func (c *Client) Resolve(ctx context.Context) error {
 	for {
 		h, released, resolved := c.peek()
 		if released {
-			return newError("cannot resolve released client")
+			return errorf("cannot resolve released client")
 		}
+
 		if resolved {
 			return nil
 		}
+
 		select {
 		case <-h.resolved:
 		case <-ctx.Done():
@@ -606,6 +608,10 @@ type ClientPromise struct {
 	h *clientHook
 }
 
+func (cp *ClientPromise) Reject(err error) {
+	cp.Fulfill(ErrorClient(err))
+}
+
 // Fulfill resolves the client promise to c.  After Fulfill returns,
 // then all future calls to the client created by NewPromisedClient will
 // be sent to c.  It is guaranteed that the hook passed to

error.go

@@ -1,49 +1,42 @@
 package capnp
 
 import (
-	"fmt"
+	"capnproto.org/go/capnp/v3/exc"
+)
 
-	"capnproto.org/go/capnp/v3/internal/errors"
+var (
+	capnperr = exc.Annotator("capnp")
 )
 
+// TODO(someday):  progressively remove exported functions and instead
+//                 rely on package 'exc'.
+
 // Unimplemented returns an error that formats as the given text and
 // will report true when passed to IsUnimplemented.
 func Unimplemented(s string) error {
-	return errors.New(errors.Unimplemented, "", s)
+	return exc.New(exc.Unimplemented, "", s)
 }
 
 // IsUnimplemented reports whether e indicates that functionality is unimplemented.
 func IsUnimplemented(e error) bool {
-	return errors.TypeOf(e) == errors.Unimplemented
+	return exc.TypeOf(e) == exc.Unimplemented
 }
 
 // Disconnected returns an error that formats as the given text and
 // will report true when passed to IsDisconnected.
 func Disconnected(s string) error {
-	return errors.New(errors.Disconnected, "", s)
+	return exc.New(exc.Disconnected, "", s)
 }
 
 // IsDisconnected reports whether e indicates a failure due to loss of a necessary capability.
 func IsDisconnected(e error) bool {
-	return errors.TypeOf(e) == errors.Disconnected
-}
-
-func newError(msg string) error {
-	return errors.New(errors.Failed, "capnp", msg)
+	return exc.TypeOf(e) == exc.Disconnected
 }
 
 func errorf(format string, args ...interface{}) error {
-	return newError(fmt.Sprintf(format, args...))
-}
-
-type annotater struct {
-	err error
-}
-
-func annotate(err error) annotater {
-	return annotater{err}
+	return capnperr.Failedf(format, args...)
 }
 
-func (a annotater) errorf(format string, args ...interface{}) error {
-	return errors.Annotate("capnp", fmt.Sprintf(format, args...), a.err)
+func annotatef(err error, format string, args ...interface{}) error {
+	return capnperr.Annotatef(err, format, args...)
 }

exc/exc.go

@@ -0,0 +1,113 @@
+// Package exc provides an error type for capnp exceptions.
+package exc
+
+import (
+	"errors"
+	"fmt"
+)
+
+// Exception is an error that designates a Cap'n Proto exception.
+type Exception struct {
+	Type   Type
+	Prefix string
+	Cause  error
+}
+
+// New creates a new error that formats as "<prefix>: <msg>".
+// The type can be recovered using the TypeOf() function.
+func New(typ Type, prefix, msg string) *Exception {
+	return &Exception{typ, prefix, errors.New(msg)}
+}
+
+func (e Exception) Error() string {
+	if e.Prefix == "" {
+		return e.Cause.Error()
+	}
+
+	return fmt.Sprintf("%s: %v", e.Prefix, e.Cause)
+}
+
+func (e Exception) Unwrap() error { return e.Cause }
+
+func (e Exception) GoString() string {
+	return fmt.Sprintf("errors.Error{Type: %s, Prefix: %q, Cause: fmt.Errorf(%q)}",
+		e.Type.GoString(),
+		e.Prefix,
+		e.Cause)
+}
+
+// Annotate is creates a new error that formats as "<prefix>: <msg>: <e>".
+// If e.Prefix == prefix, the prefix will not be duplicated.
+// The returned Error.Type == e.Type.
+func (e Exception) Annotate(prefix, msg string) *Exception {
+	if prefix != e.Prefix {
+		return &Exception{e.Type, prefix, fmt.Errorf("%s: %w", msg, e)}
+	}
+
+	return &Exception{e.Type, prefix, fmt.Errorf("%s: %w", msg, e.Cause)}
+}
+
+// Annotate creates a new error that formats as "<prefix>: <msg>: <err>".
+// If err has the same prefix, then the prefix won't be duplicated.
+// The returned error's type will match err's type.
+func Annotate(prefix, msg string, err error) *Exception {
+	if err == nil {
+		return nil
+	}
+
+	if ce, ok := err.(*Exception); ok {
+		return ce.Annotate(prefix, msg)
+	}
+
+	return &Exception{
+		Type:   Failed,
+		Prefix: prefix,
+		Cause:  fmt.Errorf("%s: %w", msg, err),
+	}
+}
+
+type Annotator string
+
+func (f Annotator) New(t Type, err error) *Exception {
+	if err == nil {
+		return nil
+	}
+
+	return &Exception{
+		Type:   t,
+		Prefix: string(f),
+		Cause:  err,
+	}
+}
+
+func (f Annotator) Failed(err error) *Exception {
+	return f.New(Failed, err)
+}
+
+func (f Annotator) Failedf(format string, args ...interface{}) *Exception {
+	return f.Failed(fmt.Errorf(format, args...))
+}
+
+func (f Annotator) Disconnected(err error) *Exception {
+	return f.New(Disconnected, err)
+}
+
+func (f Annotator) Disconnectedf(format string, args ...interface{}) *Exception {
+	return f.Disconnected(fmt.Errorf(format, args...))
+}
+
+func (f Annotator) Unimplemented(err error) *Exception {
+	return f.New(Unimplemented, err)
+}
+
+func (f Annotator) Unimplementedf(format string, args ...interface{}) *Exception {
+	return f.Unimplemented(fmt.Errorf(format, args...))
+}
+
+func (f Annotator) Annotate(err error, msg string) *Exception {
+	return Annotate(string(f), msg, err)
+}
+
+func (f Annotator) Annotatef(err error, format string, args ...interface{}) *Exception {
+	return f.Annotate(err, fmt.Sprintf(format, args...))
+}

internal/errors/errors_test.go → exc/exc_test.go

@@ -1,4 +1,4 @@
-package errors
+package exc
 
 import (
 	"errors"
@@ -13,7 +13,7 @@ func TestUnwrap(t *testing.T) {
 	var (
 		errGeneric = errors.New("something went wrong")
 		err        = Annotate("annotated", "test", errGeneric)
-		exc        Error
+		exc        = new(Exception)
 	)
 
 	assert.EqualError(t, errors.Unwrap(err), "test: something went wrong")
@@ -43,25 +43,6 @@ func TestErrorString(t *testing.T) {
 	}
 }
 
-func TestTypeOf(t *testing.T) {
-	t.Parallel()
-
-	tests := []struct {
-		err  error
-		want Type
-	}{
-		{nil, Failed},
-		{errors.New("generic"), Failed},
-		{New(Failed, "capnp", "failed error"), Failed},
-		{New(Overloaded, "capnp", "overloaded error"), Overloaded},
-		{New(Disconnected, "capnp", "disconnected error"), Disconnected},
-		{New(Unimplemented, "capnp", "unimplemented error"), Unimplemented},
-	}
-	for _, test := range tests {
-		assert.Equal(t, test.want, TypeOf(test.err))
-	}
-}
-
 func TestAnnotate(t *testing.T) {
 	t.Parallel()