Record graph error (#2195)
## Description Record information about the error Graph returned if something went wrong when fetching an item. ## Does this PR need a docs update or release note? - [ ] ✅ Yes, it's included - [ ] 🕐 Yes, but in a later PR - [x] ⛔ No ## Type of change - [ ] 🌻 Feature - [x] 🐛 Bugfix - [ ] 🗺️ Documentation - [ ] 🤖 Test - [ ] 💻 CI/Deployment - [x] 🧹 Tech Debt/Cleanup ## Issue(s) * closes #2194 ## Test Plan - [x] 💪 Manual - [ ] ⚡ Unit test - [ ] 💚 E2E
This commit is contained in:
parent
38b9fabb51
commit
5d20e30062
@ -198,14 +198,6 @@ func (col *Collection) streamItems(ctx context.Context) {
|
||||
semaphoreCh := make(chan struct{}, urlPrefetchChannelBufferSize)
|
||||
defer close(semaphoreCh)
|
||||
|
||||
updaterMu := sync.Mutex{}
|
||||
errUpdater := func(user string, err error) {
|
||||
updaterMu.Lock()
|
||||
defer updaterMu.Unlock()
|
||||
|
||||
errs = support.WrapAndAppend(user, err, errs)
|
||||
}
|
||||
|
||||
// delete all removed items
|
||||
for id := range col.removed {
|
||||
semaphoreCh <- struct{}{}
|
||||
@ -231,6 +223,14 @@ func (col *Collection) streamItems(ctx context.Context) {
|
||||
}(id)
|
||||
}
|
||||
|
||||
updaterMu := sync.Mutex{}
|
||||
errUpdater := func(user string, err error) {
|
||||
updaterMu.Lock()
|
||||
defer updaterMu.Unlock()
|
||||
|
||||
errs = support.WrapAndAppend(user, err, errs)
|
||||
}
|
||||
|
||||
// add any new items
|
||||
for id := range col.added {
|
||||
if col.ctrl.FailFast && errs != nil {
|
||||
@ -273,7 +273,7 @@ func (col *Collection) streamItems(ctx context.Context) {
|
||||
return
|
||||
}
|
||||
|
||||
errUpdater(user, err)
|
||||
errUpdater(user, support.ConnectorStackErrorTraceWrap(err, "fetching item"))
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user