Simplify Execute goroutines

core/execute.go

@@ -14,50 +14,19 @@ import (
 	"time"
 )
 
-func readStdout(
-	stdout io.ReadCloser,
-	source string,
-	postProcess func(item Item) Item,
-	items chan Item,
-	cparse chan bool,
+func readPipe(
+	f io.ReadCloser,
+	send chan []byte,
+	done chan int,
 ) {
-	var item Item
-	parseError := false
-	scanout := bufio.NewScanner(stdout)
-	for scanout.Scan() {
-		data := scanout.Bytes()
-		err := json.Unmarshal(data, &item)
-		if err != nil || item.Id == "" {
-			log.Printf("[%s: stdout] %s\n", source, strings.TrimSpace(string(data)))
-			parseError = true
-		} else {
-			if postProcess != nil {
-				item = postProcess(item)
-			}
-			item.Active = true   // These fields aren't up to
-			item.Created = 0     // the action to set and
-			item.Source = source // shouldn't be overrideable
-			log.Printf("[%s:   item] %s\n", source, item.Id)
-			items <- item
-		}
+	defer func() {
+		done <- 0
+	}()
+	scanner := bufio.NewScanner(f)
+	for scanner.Scan() {
+		data := scanner.Bytes()
+		send <- data
 	}
-	// Only send the parsing result at the end, to block main until stdout is drained
-	cparse <- parseError
-	close(items)
-}
-
-func readStderr(stderr io.ReadCloser, source string, done chan bool) {
-	scanerr := bufio.NewScanner(stderr)
-	for scanerr.Scan() {
-		text := strings.TrimSpace(scanerr.Text())
-		log.Printf("[%s: stderr] %s\n", source, text)
-	}
-	done <- true
-}
-
-func writeStdin(stdin io.WriteCloser, text string) {
-	defer stdin.Close()
-	io.WriteString(stdin, text)
 }
 
 func Execute(
@@ -68,7 +37,11 @@ func Execute(
 	input string,
 	timeout time.Duration,
 	postProcess func(item Item) Item,
-) ([]Item, []byte, error) {
+) (
+	items []Item,
+	newState []byte,
+	err error,
+) {
 	log.Printf("executing %v", argv)
 
 	if len(argv) == 0 {
@@ -115,22 +88,14 @@ func Execute(
 		return nil, nil, err
 	}
 
-	cout := make(chan Item)
-	cparse := make(chan bool)
-	cerr := make(chan bool)
-
-	// Sink routine for items produced
-	var items []Item
-	go func() {
-		for item := range cout {
-			items = append(items, item)
-		}
-	}()
+	cout := make(chan []byte)
+	cerr := make(chan []byte)
+	coutDone := make(chan int)
+	cerrDone := make(chan int)
 
 	// Routines handling the process i/o
-	go writeStdin(stdin, input)
-	go readStdout(stdout, source, postProcess, cout, cparse)
-	go readStderr(stderr, source, cerr)
+	go readPipe(stdout, cout, coutDone)
+	go readPipe(stderr, cerr, cerrDone)
 
 	// Kick off the command
 	err = cmd.Start()
@@ -138,9 +103,50 @@ func Execute(
 		return nil, nil, err
 	}
 
-	// Block until std{out,err} close
-	<-cerr
-	parseError := <-cparse
+	// Write any input to stdin and close it
+	io.WriteString(stdin, input)
+	stdin.Close()
+
+	// Collect outputs until std{out,err} close
+	parseError := false
+	stdoutDone := false
+	stderrDone := false
+monitor:
+	for {
+		select {
+		case data := <-cout:
+			var item Item
+			err := json.Unmarshal(data, &item)
+			if err != nil || item.Id == "" {
+				log.Printf("[%s: stdout] %s", source, strings.TrimSpace(string(data)))
+				parseError = true
+			} else {
+				if postProcess != nil {
+					item = postProcess(item)
+				}
+				item.Active = true   // These fields aren't up to
+				item.Created = 0     // the action to set and
+				item.Source = source // shouldn't be overrideable
+				log.Printf("[%s:   item] %s\n", source, item.Id)
+				items = append(items, item)
+			}
+
+		case data := <-cerr:
+			log.Printf("[%s: stderr] %s\n", source, data)
+
+		case <-coutDone:
+			stdoutDone = true
+			if stdoutDone && stderrDone {
+				break monitor
+			}
+
+		case <-cerrDone:
+			stderrDone = true
+			if stdoutDone && stderrDone {
+				break monitor
+			}
+		}
+	}
 
 	err = cmd.Wait()
 	if ctx.Err() == context.DeadlineExceeded {
@@ -159,12 +165,12 @@ func Execute(
 		return nil, nil, errors.New("invalid JSON")
 	}
 
-	newState, err := os.ReadFile(stateFile.Name())
+	newState, err = os.ReadFile(stateFile.Name())
 	if err != nil {
 		return nil, nil, fmt.Errorf("error: failed to read state file: %v", err)
 	}
 
-	return items, newState, nil
+	return
 }
 
 // Execute an action that takes an item as input and returns the item modified.

core/execute_test.go

@@ -6,19 +6,19 @@ import (
 )
 
 func TestExecute(t *testing.T) {
-	assertLen := func(items []Item, length int) {
+	assertLen := func(t *testing.T, items []Item, length int) {
 		t.Helper()
 		if len(items) != length {
 			t.Fatalf("Expected %d items, got %d", length, len(items))
 		}
 	}
-	assertNil := func(err error) {
+	assertNil := func(t *testing.T, err error) {
 		t.Helper()
 		if err != nil {
 			t.Fatal(err)
 		}
 	}
-	assertNotNil := func(err error) {
+	assertNotNil := func(t *testing.T, err error) {
 		t.Helper()
 		if err == nil {
 			t.Fatal("expected err")
@@ -29,138 +29,165 @@ func TestExecute(t *testing.T) {
 		return item, err
 	}
 
-	res, err := execute([]string{"true"})
-	assertNil(err)
-	assertLen(res, 0)
-
-	// Exit with error code
-	res, err = execute([]string{"false"})
-	assertNotNil(err)
-	assertLen(res, 0)
-
-	res, err = execute([]string{"sh", "-c", "exit 22"})
-	assertNotNil(err)
-	assertLen(res, 0)
-
-	// Timeout
-	res, _, err = Execute("_", []string{"sleep", "10"}, nil, nil, "", time.Millisecond, nil)
-	assertNotNil(err)
-	assertLen(res, 0)
-
-	// Returning items
-	res, err = execute([]string{"jq", "-cn", `{id: "foo"}`})
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Id != "foo" {
-		t.Fatal("jq -cn test failed")
-	}
-
-	// Read from stdin
-	res, _, err = Execute("_", []string{"jq", "-cR", `{id: .}`}, nil, nil, "bar", time.Minute, nil)
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Id != "bar" {
-		t.Fatal("jq -cR test failed")
-	}
-
-	// Set env
-	res, _, err = Execute("_", []string{"jq", "-cn", `{id: env.HELLO}`}, []string{"HELLO=baz"}, nil, "", time.Minute, nil)
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Id != "baz" {
-		t.Fatal("jq -cn env test failed")
-	}
-
-	// With logging on stderr
-	res, err = execute([]string{"sh", "-c", `echo 1>&2 Hello; jq -cn '{id: "box"}'; echo 1>&2 World`})
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Id != "box" {
-		t.Fatal("stderr test failed")
-	}
-
-	// Unsupported item field is silently discarded
-	res, err = execute([]string{"jq", "-cn", `{id: "test", unknownField: "what is this"}`})
-	assertNil(err)
-	assertLen(res, 1)
-
-	// Field with incorrect type fails
-	res, err = execute([]string{"jq", "-cn", `{id: ["list"]}`})
-	assertNotNil(err)
-	assertLen(res, 0)
-
-	res, err = execute([]string{"jq", "-cn", `{id: "test", time: "0"}`})
-	assertNotNil(err)
-	assertLen(res, 0)
-
-	res, err = execute([]string{"jq", "-cn", `{id: null}`})
-	assertNotNil(err)
-	assertLen(res, 0)
-
-	// Items with duplicate ids is not a fetch error, but it will fail to update
-	res, err = execute([]string{"jq", "-cn", `["a", "a"] | .[] | {id: .}`})
-	assertNil(err)
-	assertLen(res, 2)
-
-	// Action keys are detected even with empty values
-	res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": null}}`})
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Action["hello"] == nil {
-		t.Fatal("missing hello action")
-	}
-	if res[0].Action["goodbye"] != nil {
-		t.Fatal("nonexistent action should key to nil in Action")
-	}
-
-	res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": ""}}`})
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Action["hello"] == nil {
-		t.Fatal("missing hello action")
-	}
-
-	res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": []}}`})
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Action["hello"] == nil {
-		t.Fatal("missing hello action")
-	}
-
-	res, err = execute([]string{"jq", "-cn", `{id: "test", action: {"hello": {}}}`})
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Action["hello"] == nil {
-		t.Fatal("missing hello action")
-	}
-
-	// Read state
-	argv := []string{"sh", "-c", `cat $STATE_PATH | jq -cR '{id: "greeting", title: .} | .title = "Hello " + .title'`}
-	res, _, err = Execute("_", argv, nil, []byte("world"), "", time.Minute, nil)
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Title != "Hello world" {
-		t.Fatalf("expected 'Hello world' from read state, got '%s'", res[0].Title)
-	}
-
-	// Write state
-	argv = []string{"sh", "-c", `printf "Hello world" > $STATE_PATH; jq -cn '{id: "test"}'`}
-	res, newState, err := Execute("_", argv, nil, nil, "", time.Minute, nil)
-	assertNil(err)
-	assertLen(res, 1)
-	if string(newState) != "Hello world" {
-		t.Fatalf("expected 'Hello world' from write state, got %s", string(newState))
-	}
-
-	// Postprocessing function
-	argv = []string{"jq", "-cn", `{id: "foo"}`}
-	res, _, err = Execute("_", argv, nil, nil, "", time.Minute, func(item Item) Item {
-		item.Ttl = 123456
-		return item
+	t.Run("Noop", func(t *testing.T) {
+		res, err := execute([]string{"true"})
+		assertNil(t, err)
+		assertLen(t, res, 0)
+	})
+
+	t.Run("ExitWithErrorCode", func(t *testing.T) {
+		res, err := execute([]string{"false"})
+		assertNotNil(t, err)
+		assertLen(t, res, 0)
+	})
+
+	t.Run("ExitWithSpecificErrorCode", func(t *testing.T) {
+		res, err := execute([]string{"sh", "-c", "exit 22"})
+		assertNotNil(t, err)
+		assertLen(t, res, 0)
+	})
+
+	t.Run("Timeout", func(t *testing.T) {
+		res, _, err := Execute("_", []string{"sleep", "10"}, nil, nil, "", time.Millisecond, nil)
+		assertNotNil(t, err)
+		assertLen(t, res, 0)
+	})
+
+	t.Run("ReturningItems", func(t *testing.T) {
+		res, err := execute([]string{"jq", "-cn", `{id: "foo"}`})
+		assertNil(t, err)
+		assertLen(t, res, 1)
+		if res[0].Id != "foo" {
+			t.Fatal("jq -cn test failed")
+		}
+	})
+
+	t.Run("ReadFromStdin", func(t *testing.T) {
+		res, _, err := Execute("_", []string{"jq", "-cR", `{id: .}`}, nil, nil, "bar", time.Minute, nil)
+		assertNil(t, err)
+		assertLen(t, res, 1)
+		if res[0].Id != "bar" {
+			t.Fatal("jq -cR test failed")
+		}
+	})
+
+	t.Run("SetEnv", func(t *testing.T) {
+		res, _, err := Execute("_", []string{"jq", "-cn", `{id: env.HELLO}`}, []string{"HELLO=baz"}, nil, "", time.Minute, nil)
+		assertNil(t, err)
+		assertLen(t, res, 1)
+		if res[0].Id != "baz" {
+			t.Fatal("jq -cn env test failed")
+		}
+	})
+	t.Run("StderrLogging", func(t *testing.T) {
+		res, err := execute([]string{"sh", "-c", `echo 1>&2 Hello; jq -cn '{id: "box"}'; echo 1>&2 World`})
+		assertNil(t, err)
+		assertLen(t, res, 1)
+		if res[0].Id != "box" {
+			t.Fatal("stderr test failed")
+		}
+	})
+
+	t.Run("UnknownFieldIgnored", func(t *testing.T) {
+		res, err := execute([]string{"jq", "-cn", `{id: "test", unknownField: "what is this"}`})
+		assertNil(t, err)
+		assertLen(t, res, 1)
+	})
+
+	t.Run("IncorrectIdType", func(t *testing.T) {
+		res, err := execute([]string{"jq", "-cn", `{id: ["list"]}`})
+		assertNotNil(t, err)
+		assertLen(t, res, 0)
+	})
+
+	t.Run("IncorrectTimeType", func(t *testing.T) {
+		res, err := execute([]string{"jq", "-cn", `{id: "test", time: "0"}`})
+		assertNotNil(t, err)
+		assertLen(t, res, 0)
+	})
+
+	t.Run("NullId", func(t *testing.T) {
+		res, err := execute([]string{"jq", "-cn", `{id: null}`})
+		assertNotNil(t, err)
+		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)
+		assertLen(t, res, 2)
+	})
+
+	t.Run("ActionNullValueOk", func(t *testing.T) {
+		res, err := execute([]string{"jq", "-cn", `{id: "test", action: {"hello": null}}`})
+		assertNil(t, err)
+		assertLen(t, res, 1)
+		if res[0].Action["hello"] == nil {
+			t.Fatal("missing hello action")
+		}
+		if res[0].Action["goodbye"] != nil {
+			t.Fatal("nonexistent action should key to nil in Action")
+		}
+	})
+
+	t.Run("ActionEmptyStringOk", func(t *testing.T) {
+		res, err := execute([]string{"jq", "-cn", `{id: "test", action: {"hello": ""}}`})
+		assertNil(t, err)
+		assertLen(t, res, 1)
+		if res[0].Action["hello"] == nil {
+			t.Fatal("missing hello action")
+		}
+	})
+
+	t.Run("ActionEmptyArrayOk", func(t *testing.T) {
+		res, err := execute([]string{"jq", "-cn", `{id: "test", action: {"hello": []}}`})
+		assertNil(t, err)
+		assertLen(t, res, 1)
+		if res[0].Action["hello"] == nil {
+			t.Fatal("missing hello action")
+		}
+	})
+
+	t.Run("ActionEmptyObjectOk", func(t *testing.T) {
+		res, err := execute([]string{"jq", "-cn", `{id: "test", action: {"hello": {}}}`})
+		assertNil(t, err)
+		assertLen(t, res, 1)
+		if res[0].Action["hello"] == nil {
+			t.Fatal("missing hello action")
+		}
+	})
+
+	t.Run("ReadState", func(t *testing.T) {
+		argv := []string{"sh", "-c", `cat $STATE_PATH | jq -cR '{id: "greeting", title: .} | .title = "Hello " + .title'`}
+		res, _, err := Execute("_", argv, nil, []byte("world"), "", time.Minute, nil)
+		assertNil(t, err)
+		assertLen(t, res, 1)
+		if res[0].Title != "Hello world" {
+			t.Fatalf("expected 'Hello world' from read state, got '%s'", res[0].Title)
+		}
+	})
+
+	t.Run("WriteState", func(t *testing.T) {
+		argv := []string{"sh", "-c", `printf "Hello world" > $STATE_PATH; jq -cn '{id: "test"}'`}
+		res, newState, err := Execute("_", argv, nil, nil, "", time.Minute, nil)
+		assertNil(t, err)
+		assertLen(t, res, 1)
+		if string(newState) != "Hello world" {
+			t.Fatalf("expected 'Hello world' from write state, got %s", string(newState))
+		}
+	})
+
+	t.Run("PostprocessSetTtl", func(t *testing.T) {
+		argv := []string{"jq", "-cn", `{id: "foo"}`}
+		res, _, err := Execute("_", argv, nil, nil, "", time.Minute, func(item Item) Item {
+			item.Ttl = 123456
+			return item
+		})
+		assertNil(t, err)
+		assertLen(t, res, 1)
+		if res[0].Ttl != 123456 {
+			t.Fatalf("expected ttl to be set to 123456, got %d", res[0].Ttl)
+		}
 	})
-	assertNil(err)
-	assertLen(res, 1)
-	if res[0].Ttl != 123456 {
-		t.Fatalf("expected ttl to be set to 123456, got %d", res[0].Ttl)
-	}
 }

core/source_test.go

@@ -183,7 +183,7 @@ func TestOnCreateAction(t *testing.T) {
 	items := execute([]string{"jq", "-cn", `{id: "one"}`})
 	add, _, err := update(items)
 	if add != 1 || err != nil {
-		t.Fatal("failed update with noop oncreate")
+		t.Fatalf("failed update with noop oncreate: %v", err)
 	}
 	updated := getItem("one")
 	updated.Created = 0 // zero out for comparison with pre-insert item