checks

series/collector.go

@@ -118,7 +118,7 @@ func NewCollector(ctx context.Context, opts ...CollectorOptionProvider) (*Collec
 }
 
 func (c *Collector) Close() error {
-	c.cancel()
+	ft.SafeCall(c.cancel)
 	if c.broker != nil {
 		c.broker.Stop()
 	}

series/persistance.go

@@ -382,6 +382,7 @@ func SocketBackend(opts ...CollectorBakendSocketOptionProvider) (CollectorBacken
 		return iter.ProcessParallel(
 			func(ctx context.Context, pub MetricPublisher) (err error) {
 				defer func() { grip.Error(err); grip.Infoln("done worker;", counter.Load()) }()
+				counter.Add(1)
 				grip.Info("start worker")
 				conn, err := connCache.WaitFront(ctx)
 				grip.Info("got first conn")
@@ -441,7 +442,7 @@ func SocketBackend(opts ...CollectorBakendSocketOptionProvider) (CollectorBacken
 			},
 			fun.WorkerGroupConfErrorHandler(ec.Add),
 			fun.WorkerGroupConfNumWorkers(conf.MessageWorkers),
-			conf.MessageErrorHandling.poolErrorOptions(),
+			// conf.MessageErrorHandling.poolErrorOptions(),
 		).PreHook(conPoolWorker.Operation(ec.Add).Once().Go()).PostHook(func() { /* do cancel */ }).Run(ctx)
 	}, nil
 }

series/series_test.go

@@ -34,9 +34,8 @@ func TestIntegration(t *testing.T) {
 					CollectorBackendSocketConfMinDialRetryDelay(100*time.Millisecond),
 					CollectorBackendSocketConfIdleConns(6),
 					CollectorBackendSocketConfMaxDialRetryDelay(time.Second),
-					CollectorBackendSocketConfMessageErrorHandling(CollectorBackendSocketErrorCollect),
-					CollectorBackendSocketConfDialErrorHandling(CollectorBackendSocketErrorCollect),
-				),
+					CollectorBackendSocketConfMessageErrorHandling(CollectorBackendSocketErrorPanic),
+					CollectorBackendSocketConfDialErrorHandling(CollectorBackendSocketErrorPanic)),
 				),
 			))
 		assert.NotError(t, err)