Error handling in ListExisting(Received)Shares

go.mod

@@ -5,6 +5,7 @@ require (
 	github.com/CiscoM31/godata v1.0.8
 	github.com/Masterminds/sprig v2.22.0+incompatible
 	github.com/ReneKroon/ttlcache/v2 v2.11.0
+	github.com/alitto/pond v1.9.2
 	github.com/beevik/etree v1.4.1
 	github.com/bluele/gcache v0.0.2
 	github.com/c-bata/go-prompt v0.2.6
@@ -41,8 +42,8 @@ require (
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/nats-io/nats.go v1.37.0
 	github.com/onsi/ginkgo v1.16.5
-	github.com/owncloud/libre-graph-api-go v1.0.5-0.20240425090020-dba6d1507c38
 	github.com/onsi/gomega v1.36.2
+	github.com/owncloud/libre-graph-api-go v1.0.5-0.20240425090020-dba6d1507c38
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.20.5
 	github.com/rs/cors v1.11.1
@@ -73,7 +74,7 @@ require (
 	github.com/Azure/go-ntlmssp v0.0.0-20221128193559-754e69321358 // indirect
 	github.com/Masterminds/goutils v1.1.1 // indirect
 	github.com/Masterminds/semver v1.5.0 // indirect
-	github.com/alitto/pond v1.8.3
+	github.com/alitto/pond/v2 v2.1.6
 	github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bmizerany/pat v0.0.0-20210406213842-e4b6760bdd6f // indirect

go.sum

@@ -804,8 +804,10 @@ github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk5
 github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137/go.mod h1:OMCwj8VM1Kc9e19TLln2VL61YJF0x1XFtfdL4JdbSyE=
 github.com/alexbrainman/sspi v0.0.0-20231016080023-1a75b4708caa h1:LHTHcTQiSGT7VVbI0o4wBRNQIgn917usHWOd6VAffYI=
 github.com/alexbrainman/sspi v0.0.0-20231016080023-1a75b4708caa/go.mod h1:cEWa1LVoE5KvSD9ONXsZrj0z6KqySlCCNKHlLzbqAt4=
-github.com/alitto/pond v1.8.3 h1:ydIqygCLVPqIX/USe5EaV/aSRXTRXDEI9JwuDdu+/xs=
-github.com/alitto/pond v1.8.3/go.mod h1:CmvIIGd5jKLasGI3D87qDkQxjzChdKMmnXMg3fG6M6Q=
+github.com/alitto/pond v1.9.2 h1:9Qb75z/scEZVCoSU+osVmQ0I0JOeLfdTDafrbcJ8CLs=
+github.com/alitto/pond v1.9.2/go.mod h1:xQn3P/sHTYcU/1BR3i86IGIrilcrGC2LiS+E2+CJWsI=
+github.com/alitto/pond/v2 v2.1.6 h1:6U3nSOjxpuNyvjIKjjRkpS2JDdgX5JqBm9GO2urcCjM=
+github.com/alitto/pond/v2 v2.1.6/go.mod h1:xkjYEgQ05RSpWdfSd1nM3OVv7TBhLdy7rMp3+2Nq+yE=
 github.com/andybalholm/brotli v1.0.4/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/apache/arrow/go/v10 v10.0.1/go.mod h1:YvhnlEePVnBS4+0z3fhPfUy7W1Ikj0Ih0vcRo/gZ1M0=

internal/grpc/services/gateway/usershareprovider.go

@@ -23,7 +23,7 @@ import (
 	"fmt"
 	"path"
 
-	"github.com/alitto/pond"
+	"github.com/alitto/pond/v2"
 	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
 	userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
 	rpc "github.com/cs3org/go-cs3apis/cs3/rpc/v1beta1"
@@ -204,28 +204,29 @@ func (s *svc) ListExistingShares(ctx context.Context, req *collaboration.ListSha
 	}
 
 	sharesCh := make(chan *gateway.ShareResourceInfo, len(shares.Shares))
-	pool := pond.New(50, len(shares.Shares))
+	pool := pond.NewPool(50)
+	// TODO(lopresti) incorporate the cache layer from internal/http/services/owncloud/ocs/handlers/apps/sharing/shares/shares.go
+
 	for _, share := range shares.Shares {
 		share := share
-		// TODO (gdelmont): we should report any eventual error raised by the goroutines
-		pool.Submit(func() {
-			// TODO(lopresti) incorporate the cache layer from internal/http/services/owncloud/ocs/handlers/apps/sharing/shares/shares.go
+		pool.SubmitErr(func() error {
 			stat, err := s.Stat(ctx, &provider.StatRequest{
 				Ref: &provider.Reference{
 					ResourceId: share.ResourceId,
 				},
 			})
 			if err != nil {
-				return
+				return err
 			}
 			if stat.Status.Code != rpc.Code_CODE_OK {
-				return
+				return errors.New("An error occurred: " + stat.Status.Message)
 			}
 
 			sharesCh <- &gateway.ShareResourceInfo{
 				ResourceInfo: stat.Info,
 				Share:        share,
 			}
+			return nil
 		})
 	}
 
@@ -237,11 +238,18 @@ func (s *svc) ListExistingShares(ctx context.Context, req *collaboration.ListSha
 		}
 		done <- struct{}{}
 	}()
-	pool.StopAndWait()
+	err = pool.Stop().Wait()
 	close(sharesCh)
 	<-done
 	close(done)
 
+	if err != nil {
+		return &gateway.ListExistingSharesResponse{
+			ShareInfos: sris,
+			Status:     status.NewInternal(ctx, err, "An error occured listing existing shares"),
+		}, err
+	}
+
 	return &gateway.ListExistingSharesResponse{
 		ShareInfos: sris,
 		Status:     status.NewOK(ctx),
@@ -318,13 +326,12 @@ func (s *svc) ListExistingReceivedShares(ctx context.Context, req *collaboration
 	}
 
 	sharesCh := make(chan *gateway.ReceivedShareResourceInfo, len(rshares.Shares))
-	pool := pond.New(50, len(rshares.Shares))
+	pool := pond.NewPool(50)
 	for _, rs := range rshares.Shares {
 		rs := rs
-		// TODO (gdelmont): we should report any eventual error raised by the goroutines
-		pool.Submit(func() {
+		pool.SubmitErr(func() error {
 			if rs.State == collaboration.ShareState_SHARE_STATE_REJECTED || rs.State == collaboration.ShareState_SHARE_STATE_INVALID {
-				return
+				return errors.New("Invalid Share State")
 			}
 
 			// TODO(lopresti) incorporate the cache layer from internal/http/services/owncloud/ocs/handlers/apps/sharing/shares/shares.go
@@ -334,16 +341,17 @@ func (s *svc) ListExistingReceivedShares(ctx context.Context, req *collaboration
 				},
 			})
 			if err != nil {
-				return
+				return err
 			}
 			if stat.Status.Code != rpc.Code_CODE_OK {
-				return
+				return errors.New("An error occurred: " + stat.Status.Message)
 			}
 
 			sharesCh <- &gateway.ReceivedShareResourceInfo{
 				ResourceInfo:  stat.Info,
 				ReceivedShare: rs,
 			}
+			return nil
 		})
 	}
 
@@ -355,11 +363,18 @@ func (s *svc) ListExistingReceivedShares(ctx context.Context, req *collaboration
 		}
 		done <- struct{}{}
 	}()
-	pool.StopAndWait()
+	err = pool.Stop().Wait()
 	close(sharesCh)
 	<-done
 	close(done)
 
+	if err != nil {
+		return &gateway.ListExistingReceivedSharesResponse{
+			ShareInfos: sris,
+			Status:     status.NewInternal(ctx, err, "An error occured listing received shares"),
+		}, err
+	}
+
 	return &gateway.ListExistingReceivedSharesResponse{
 		ShareInfos: sris,
 		Status:     status.NewOK(ctx),