Fixes

caseydavenport · caseydavenport · commit 522e2d5b3ab1 · 2025-04-15T20:39:46.000-07:00
diff --git a/goldmane/pkg/storage/bucket.go b/goldmane/pkg/storage/bucket.go
@@ -129,7 +129,7 @@ func (b *AggregationBucket) Iter(fn func(FlowBuilder) bool) {
 	defer b.Unlock()
 
 	b.Flows.Iter(func(d *DiachronicFlow) error {
-		if !fn(NewDeferredFlowBuilder(d, b.StartTime, b.EndTime)) {
+		if fn(NewDeferredFlowBuilder(d, b.StartTime, b.EndTime)) {
 			return set.StopIteration
 		}
 		return nil
diff --git a/goldmane/pkg/stream/stream.go b/goldmane/pkg/stream/stream.go
@@ -73,6 +73,7 @@ func (s *stream) run() {
 			b.Iter(func(f storage.FlowBuilder) bool {
 				if err := chanutil.WriteWithDeadline(s.ctx, s.out, f, 60*time.Second); err != nil {
 					// If we hit an error, indicate that we should stop iteration.
+					s.rl.WithFields(logrus.Fields{"id": s.ID}).WithError(err).Debug("Error writing flow to stream output")
 					return true
 				}
 				// If we didn't hit an error, continue iteration.

Original file line number	Diff line number	Diff line change
`@@ -129,7 +129,7 @@ func (b *AggregationBucket) Iter(fn func(FlowBuilder) bool) {`
`129`	`129`	`defer b.Unlock()`
`130`	`130`
`131`	`131`	`b.Flows.Iter(func(d *DiachronicFlow) error {`
`132`		`- if !fn(NewDeferredFlowBuilder(d, b.StartTime, b.EndTime)) {`
	`132`	`+ if fn(NewDeferredFlowBuilder(d, b.StartTime, b.EndTime)) {`
`133`	`133`	`return set.StopIteration`
`134`	`134`	`}`
`135`	`135`	`return nil`
Original file line number	Diff line number	Diff line change
`@@ -73,6 +73,7 @@ func (s *stream) run() {`
`73`	`73`	`b.Iter(func(f storage.FlowBuilder) bool {`
`74`	`74`	`if err := chanutil.WriteWithDeadline(s.ctx, s.out, f, 60*time.Second); err != nil {`
`75`	`75`	`// If we hit an error, indicate that we should stop iteration.`
	`76`	`+ s.rl.WithFields(logrus.Fields{"id": s.ID}).WithError(err).Debug("Error writing flow to stream output")`
`76`	`77`	`return true`
`77`	`78`	`}`
`78`	`79`	`// If we didn't hit an error, continue iteration.`