Skip to content

Commit 2f7289b

Browse files
committed
Use map for accumulating errors
1 parent 467f55f commit 2f7289b

File tree

1 file changed

+22
-15
lines changed

1 file changed

+22
-15
lines changed

pkg/agent/core/ngt/handler/grpc/object.go

Lines changed: 22 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,6 @@ import (
2828
"github.com/vdaas/vald/internal/net/grpc/errdetails"
2929
"github.com/vdaas/vald/internal/net/grpc/status"
3030
"github.com/vdaas/vald/internal/observability/trace"
31-
"github.com/vdaas/vald/internal/slices"
3231
)
3332

3433
func (s *server) Exists(ctx context.Context, uid *payload.Object_ID) (res *payload.Object_ID, err error) {
@@ -211,8 +210,9 @@ func (s *server) StreamListObject(_ *payload.Object_List_Request, stream vald.Ob
211210

212211
var (
213212
mu sync.Mutex
214-
emu sync.Mutex
213+
emu sync.RWMutex
215214
errs = make([]error, 0, s.ngt.Len())
215+
emap = make(map[string]struct{})
216216
)
217217
s.ngt.ListObjectFunc(ctx, func(uuid string, _ uint32, _ int64) bool {
218218
vec, ts, err := s.ngt.GetObject(uuid)
@@ -241,27 +241,34 @@ func (s *server) StreamListObject(_ *payload.Object_List_Request, stream vald.Ob
241241
mu.Unlock()
242242

243243
if err != nil {
244-
emu.Lock()
245-
errs = append(errs, err)
246-
emu.Unlock()
244+
emu.RLock()
245+
_, ok := emap[err.Error()]
246+
emu.RUnlock()
247+
if !ok {
248+
emu.Lock()
249+
errs = append(errs, err)
250+
emap[err.Error()] = struct{}{}
251+
emu.Unlock()
252+
}
247253
}
248254

249255
// always return true to continue streaming and let the context cancel the Range when stream closes.
250256
return true
251257
})
252258

253259
if len(errs) != 0 {
254-
slices.RemoveDuplicates(errs, func(left, right error) bool {
255-
return left.Error() < right.Error()
256-
})
257-
err = errors.Join(errs...)
258-
st, msg, err := status.ParseError(err, codes.Internal, "failed to parse StreamListObject final gRPC error response")
259-
if span != nil {
260-
span.RecordError(err)
261-
span.SetAttributes(trace.FromGRPCStatus(st.Code(), msg)...)
262-
span.SetStatus(trace.StatusError, msg)
260+
// Register all the gRPC codes to the span. Doing this because ParseError cannot parse joined error.
261+
for _, e := range errs {
262+
st, msg, err := status.ParseError(e, codes.Internal, "failed to parse StreamListObject final gRPC error response")
263+
if span != nil {
264+
span.RecordError(err)
265+
span.SetAttributes(trace.FromGRPCStatus(st.Code(), msg)...)
266+
span.SetStatus(trace.StatusError, msg)
267+
}
263268
}
264-
return err
269+
270+
// now join all the errors to return
271+
return errors.Join(errs...)
265272
}
266273
return nil
267274
}

0 commit comments

Comments
 (0)