Fail execution on duplicate item ids

core/execute.go

@@ -154,6 +154,8 @@ func Execute(
 	parseError := false
 	stdoutDone := false
 	stderrDone := false
+	duplicateItem := ""
+	itemIds := make(map[string]bool)
 monitor:
 	for {
 		select {
@@ -166,6 +168,16 @@ monitor:
 				logs = append(logs, msg)
 				parseError = true
 			} else {
+				if itemIds[item.Id] {
+					msg := fmt.Sprintf("[%s:   item] %s (duplicate)", source, item.Id)
+					log.Print(msg)
+					logs = append(logs, msg)
+					duplicateItem = item.Id
+					cmd.Cancel()
+					break monitor
+				}
+				itemIds[item.Id] = true
+
 				if postProcess != nil {
 					item = postProcess(item)
 				}
@@ -198,7 +210,12 @@ monitor:
 	}
 
 	err = cmd.Wait()
-	if ctx.Err() == context.DeadlineExceeded {
+	if duplicateItem != "" {
+		err = fmt.Errorf("returned item %s twice: %v", duplicateItem, err)
+		errItem = makeErrorItem(err, logs)
+		log.Printf("error: %v", err)
+		return nil, nil, errItem, err
+	} else if ctx.Err() == context.DeadlineExceeded {
 		err = fmt.Errorf("timed out after %v", timeout)
 		errItem = makeErrorItem(err, logs)
 		log.Printf("error: %v", err)

core/execute_test.go

@@ -113,11 +113,10 @@ func TestExecute(t *testing.T) {
 		assertLen(t, res, 0)
 	})
 
-	// TODO maybe this *should* be an Execute error, via a map[string]bool
 	t.Run("DuplicateItemIds", func(t *testing.T) {
 		res, err := execute([]string{"jq", "-cn", `["a", "a"] | .[] | {id: .}`})
-		assertNil(t, err)
+		assertNotNil(t, err)
-		assertLen(t, res, 2)
+		assertLen(t, res, 0)
 	})
 
 	t.Run("ActionNullValueOk", func(t *testing.T) {