diff --git a/pkg/poller/inputs.go b/pkg/poller/inputs.go
index 764f7672..1546a4f5 100644
--- a/pkg/poller/inputs.go
+++ b/pkg/poller/inputs.go
@@ -65,6 +65,21 @@ func NewInput(i *InputPlugin) {
 	inputs = append(inputs, i)
 }
 
+// recoverInitialize runs input.Initialize and converts a panic into an error.
+// Each input plugin's Initialize call runs in its own goroutine, so a panic
+// there cannot be caught by a recover() in the caller; left unrecovered, it
+// crashes the whole process before the app even starts.
+// See https://github.com/unpoller/unpoller/issues/1030
+func recoverInitialize(input *InputPlugin, l Logger) (err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			err = fmt.Errorf("input plugin %s panicked initializing: %v", input.Name, r) //nolint:err113
+		}
+	}()
+
+	return input.Initialize(l)
+}
+
 // InitializeInputs runs the passed-in initializer method for each input plugin.
 func (u *UnifiPoller) InitializeInputs() error {
 	inputSync.RLock()
@@ -82,28 +97,12 @@ func (u *UnifiPoller) InitializeInputs() error {
 		go func(input *InputPlugin) {
 			defer wg.Done()
 
-			sent := false
-
-			// A panicking input plugin runs in its own goroutine, so it cannot be
-			// caught by a recover() in the caller. Without this, a panic here
-			// crashes the whole process before the app even starts.
-			// See https://github.com/unpoller/unpoller/issues/1030
-			defer func() {
-				if r := recover(); r != nil && !sent {
-					u.LogErrorf("input plugin %s panicked initializing (see issue #1030): %v", input.Name, r)
-
-					errChan <- fmt.Errorf("input plugin %s panicked initializing: %v", input.Name, r) //nolint:err113
-				}
-			}()
-
 			// This must return, or the app locks up here.
 			u.LogDebugf("inititalizing input... %s", input.Name)
 
-			if err := input.Initialize(u); err != nil {
+			if err := recoverInitialize(input, u); err != nil {
 				u.LogDebugf("error initializing input ... %s", input.Name)
 
-				sent = true
-
 				errChan <- err
 
 				return
@@ -111,8 +110,6 @@ func (u *UnifiPoller) InitializeInputs() error {
 
 			u.LogDebugf("input successfully initialized ... %s", input.Name)
 
-			sent = true
-
 			errChan <- nil
 		}(input)
 	}
@@ -149,6 +146,18 @@ type eventInputResult struct {
 	err  error
 }
 
+// recoverEvents runs input.Events and converts a panic into an error. See
+// recoverInitialize for why this is necessary.
+func recoverEvents(input *InputPlugin, filter *Filter) (e *Events, err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			err = fmt.Errorf("input plugin %s panicked collecting events: %v", input.Name, r) //nolint:err113
+		}
+	}()
+
+	return input.Events(filter)
+}
+
 func collectEvents(filter *Filter, inputs []*InputPlugin) (*Events, error) {
 	resultChan := make(chan eventInputResult, len(inputs))
 	wg := &sync.WaitGroup{}
@@ -159,41 +168,21 @@ func collectEvents(filter *Filter, inputs []*InputPlugin) (*Events, error) {
 		go func(input *InputPlugin) {
 			defer wg.Done()
 
-			sent := false
-
-			// A panicking input plugin runs in its own goroutine, so it cannot be
-			// caught by a recover() in the caller. Without this, a panic here
-			// (e.g. from a malformed controller response) crashes the whole
-			// process. See https://github.com/unpoller/unpoller/issues/1030
-			defer func() {
-				if r := recover(); r != nil && !sent {
-					resultChan <- eventInputResult{
-						err: fmt.Errorf("input plugin %s panicked collecting events (see issue #1030): %v", input.Name, r),
-					}
-				}
-			}()
-
 			if filter != nil &&
 				filter.Name != "" &&
 				!strings.EqualFold(input.Name, filter.Name) {
-				sent = true
-
 				resultChan <- eventInputResult{}
 
 				return
 			}
 
-			e, err := input.Events(filter)
+			e, err := recoverEvents(input, filter)
 			if err != nil {
-				sent = true
-
 				resultChan <- eventInputResult{err: err}
 
 				return
 			}
 
-			sent = true
-
 			resultChan <- eventInputResult{logs: e.Logs}
 		}(input)
 	}
@@ -238,6 +227,20 @@ type metricInputResult struct {
 	err    error
 }
 
+// recoverMetrics runs input.Metrics and converts a panic into an error (e.g.
+// from a malformed controller response, such as an unexpected Site Speed
+// Test aggregated-dashboard payload). See recoverInitialize for why this is
+// necessary.
+func recoverMetrics(input *InputPlugin, filter *Filter) (m *Metrics, err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			err = fmt.Errorf("input plugin %s panicked collecting metrics: %v", input.Name, r) //nolint:err113
+		}
+	}()
+
+	return input.Metrics(filter)
+}
+
 func collectMetrics(filter *Filter, inputs []*InputPlugin) (*Metrics, error) {
 	resultChan := make(chan metricInputResult, len(inputs))
 	wg := &sync.WaitGroup{}
@@ -248,34 +251,15 @@ func collectMetrics(filter *Filter, inputs []*InputPlugin) (*Metrics, error) {
 		go func(input *InputPlugin) {
 			defer wg.Done()
 
-			sent := false
-
-			// A panicking input plugin runs in its own goroutine, so it cannot be
-			// caught by a recover() in the caller. Without this, a panic here
-			// (e.g. from a malformed controller response, such as an unexpected
-			// Site Speed Test aggregated-dashboard payload) crashes the whole
-			// process. See https://github.com/unpoller/unpoller/issues/1030
-			defer func() {
-				if r := recover(); r != nil && !sent {
-					resultChan <- metricInputResult{
-						err: fmt.Errorf("input plugin %s panicked collecting metrics (see issue #1030): %v", input.Name, r),
-					}
-				}
-			}()
-
 			if filter != nil &&
 				filter.Name != "" &&
 				!strings.EqualFold(input.Name, filter.Name) {
-				sent = true
-
 				resultChan <- metricInputResult{}
 
 				return
 			}
 
-			m, err := input.Metrics(filter)
-			sent = true
-
+			m, err := recoverMetrics(input, filter)
 			resultChan <- metricInputResult{metric: m, err: err}
 		}(input)
 	}