refactor(pubsublite): add context to error messages for leaf services (…

…#3649)

To disambiguate error messages from leaf services when they are propagated to the top-level publisher and subscriber clients.

pubsublite/internal/test/util.go

@@ -18,13 +18,14 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
+	"golang.org/x/xerrors"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
 
 // ErrorEqual compares two errors for equivalence.
 func ErrorEqual(got, want error) bool {
-	if got == want {
+	if xerrors.Is(got, want) {
 		return true
 	}
 	return cmp.Equal(got, want, cmpopts.EquateErrors())

pubsublite/internal/wire/assigner.go

@@ -63,6 +63,7 @@ type partitionAssignmentReceiver func(partitionSet) error
 type assigner struct {
 	// Immutable after creation.
 	assignmentClient  *vkit.PartitionAssignmentClient
+	subscription      string
 	initialReq        *pb.PartitionAssignmentRequest
 	receiveAssignment partitionAssignmentReceiver
 	metadata          pubsubMetadata
@@ -81,6 +82,7 @@ func newAssigner(ctx context.Context, assignmentClient *vkit.PartitionAssignment
 
 	a := &assigner{
 		assignmentClient: assignmentClient,
+		subscription:     subscriptionPath,
 		initialReq: &pb.PartitionAssignmentRequest{
 			Request: &pb.PartitionAssignmentRequest_Initial{
 				Initial: &pb.InitialPartitionAssignmentRequest{
@@ -164,7 +166,7 @@ func (a *assigner) handleAssignment(assignment *pb.PartitionAssignment) error {
 }
 
 func (a *assigner) unsafeInitiateShutdown(targetStatus serviceStatus, err error) {
-	if !a.unsafeUpdateStatus(targetStatus, err) {
+	if !a.unsafeUpdateStatus(targetStatus, wrapError("assigner", a.subscription, err)) {
 		return
 	}
 	// No data to send. Immediately terminate the stream.

pubsublite/internal/wire/committer.go

@@ -42,6 +42,7 @@ const commitCursorPeriod = 50 * time.Millisecond
 type committer struct {
 	// Immutable after creation.
 	cursorClient *vkit.CursorClient
+	subscription subscriptionPartition
 	initialReq   *pb.StreamingCommitCursorRequest
 	metadata     pubsubMetadata
 
@@ -59,6 +60,7 @@ func newCommitter(ctx context.Context, cursor *vkit.CursorClient, settings Recei
 
 	c := &committer{
 		cursorClient: cursor,
+		subscription: subscription,
 		initialReq: &pb.StreamingCommitCursorRequest{
 			Request: &pb.StreamingCommitCursorRequest_Initial{
 				Initial: &pb.InitialCommitCursorRequest{
@@ -201,7 +203,7 @@ func (c *committer) unsafeCommitOffsetToStream() {
 }
 
 func (c *committer) unsafeInitiateShutdown(targetStatus serviceStatus, err error) {
-	if !c.unsafeUpdateStatus(targetStatus, err) {
+	if !c.unsafeUpdateStatus(targetStatus, wrapError("committer", c.subscription.String(), err)) {
 		return
 	}
 

pubsublite/internal/wire/errors.go

@@ -16,6 +16,8 @@ package wire
 import (
 	"errors"
 	"fmt"
+
+	"golang.org/x/xerrors"
 )
 
 // Errors exported from this package.
@@ -41,3 +43,10 @@ var (
 	// stopping.
 	ErrServiceStopped = errors.New("pubsublite: service has stopped or is stopping")
 )
+
+func wrapError(context, resource string, err error) error {
+	if err != nil {
+		return xerrors.Errorf("%s(%s): %w", context, resource, err)
+	}
+	return err
+}

pubsublite/internal/wire/publisher.go

@@ -236,7 +236,7 @@ func (pp *singlePartitionPublisher) onResponse(response interface{}) {
 //
 // Expected to be called with singlePartitionPublisher.mu held.
 func (pp *singlePartitionPublisher) unsafeInitiateShutdown(targetStatus serviceStatus, err error) {
-	if !pp.unsafeUpdateStatus(targetStatus, err) {
+	if !pp.unsafeUpdateStatus(targetStatus, wrapError("publisher", pp.topic.String(), err)) {
 		return
 	}
 

pubsublite/internal/wire/resources.go

@@ -149,7 +149,15 @@ type topicPartition struct {
 	Partition int
 }
 
+func (tp topicPartition) String() string {
+	return fmt.Sprintf("%s/partitions/%d", tp.Path, tp.Partition)
+}
+
 type subscriptionPartition struct {
 	Path      string
 	Partition int
 }
+
+func (sp subscriptionPartition) String() string {
+	return fmt.Sprintf("%s/partitions/%d", sp.Path, sp.Partition)
+}

pubsublite/internal/wire/subscriber.go

@@ -325,7 +325,7 @@ func (s *subscribeStream) unsafeSendFlowControl(req *pb.FlowControlRequest) {
 }
 
 func (s *subscribeStream) unsafeInitiateShutdown(targetStatus serviceStatus, err error) {
-	if !s.unsafeUpdateStatus(targetStatus, err) {
+	if !s.unsafeUpdateStatus(targetStatus, wrapError("subscriber", s.subscription.String(), err)) {
 		return
 	}