fix data race in config/schema.go (#432)

Author: miparnisari

.github/workflows/test.yml

@@ -13,7 +13,7 @@ jobs:
     if: ${{ github.repository == 'warpstreamlabs/bento' || github.event_name != 'schedule' }}
     runs-on: ubuntu-latest
     env:
-      CGO_ENABLED: 0
+      CGO_ENABLED: 1
     steps:
 
     - name: Checkout code

Makefile

@@ -1,5 +1,5 @@
 .PHONY: all serverless deps docker docker-cgo clean docs test test-race test-integration fmt lint install deploy-docs playground
-
+.DEFAULT_GOAL := help
 TAGS ?=
 
 GOMAXPROCS         ?= 1
@@ -29,12 +29,15 @@ DOCS_FLAGS ?=
 APPS = bento
 all: $(APPS)
 
-install: $(APPS)
+help: ## Show this help
+	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}'
+
+install: $(APPS) ## Install binaries to $(INSTALL_DIR)
 	@install -d $(INSTALL_DIR)
 	@rm -f $(INSTALL_DIR)/bento
 	@cp $(PATHINSTBIN)/* $(INSTALL_DIR)/
 
-deps:
+deps: ## Go mod tidy
 	@go mod tidy
 
 SOURCE_FILES = $(shell find internal public cmd -type f)
@@ -85,16 +88,16 @@ fmt:
 	@go list -f {{.Dir}} ./... | xargs -I{} goimports -w -local github.com/warpstreamlabs/bento {}
 	@go mod tidy
 
-lint:
+lint: ## Run Go linter
 	@go vet $(GO_FLAGS) ./...
 	@golangci-lint -j $(GOMAXPROCS) run --fix -c .golangci.yml ./...
 
-test: $(APPS)
+test: $(APPS) ## Run tests
 	@go test $(GO_FLAGS) -ldflags "$(LD_FLAGS)" -timeout 3m ./...
 	@$(PATHINSTBIN)/bento template lint $(TEMPLATE_FILES)
 	@$(PATHINSTBIN)/bento test ./config/test/...
 
-test-race: $(APPS)
+test-race: $(APPS) ## Run tests with -race
 	@go test $(GO_FLAGS) -ldflags "$(LD_FLAGS)" -timeout 3m -race ./...
 
 test-integration:

internal/config/schema.go

@@ -45,8 +45,6 @@ func (t *Type) GetRawSource() any {
 	return t.rawSource
 }
 
-var httpField = docs.FieldObject(fieldHTTP, "Configures the service-wide HTTP server.").WithChildren(api.Spec()...)
-
 func observabilityFields() docs.FieldSpecs {
 	defaultMetrics := "none"
 	if _, exists := bundle.GlobalEnvironment.GetDocs("prometheus", docs.TypeMetrics); exists {
@@ -74,6 +72,8 @@ func errorHandlingFields() docs.FieldSpecs {
 
 // Spec returns a docs.FieldSpec for an entire Bento configuration.
 func Spec() docs.FieldSpecs {
+	var httpField = docs.FieldObject(fieldHTTP, "Configures the service-wide HTTP server.").WithChildren(api.Spec()...)
+
 	fields := docs.FieldSpecs{httpField}
 	fields = append(fields, stream.Spec()...)
 	fields = append(fields, manager.Spec()...)

internal/config/schema_test.go

@@ -0,0 +1,26 @@
+package config
+
+import (
+	"sync"
+	"testing"
+)
+
+func TestSpecIsThreadSafe(t *testing.T) {
+	t.Parallel()
+
+	// use a loop to increase the chance of reproducing the data race
+	for i := 0; i < 100; i++ {
+		var wg sync.WaitGroup
+		goroutines := 2
+		wg.Add(goroutines)
+		for j := 0; j < goroutines; j++ {
+			go func() {
+				defer wg.Done()
+				newSpec := Spec()
+				newSpec.SetDefault(false, "http", "enabled")
+			}()
+		}
+
+		wg.Wait()
+	}
+}

internal/impl/pure/input_inproc_test.go

@@ -37,7 +37,7 @@ inproc: foo
 }
 
 func TestInprocDryRunNoConn(t *testing.T) {
-	ctx, done := context.WithTimeout(context.Background(), time.Second*30)
+	ctx, done := context.WithTimeout(context.Background(), time.Second*60)
 	defer done()
 
 	t.Parallel()

internal/impl/pure/processor_workflow_test.go

@@ -1028,15 +1028,21 @@ workflow:
 	strm, tracer, err := strmBuilder.BuildTraced()
 	require.NoError(t, err)
 
-	tCtx, done := context.WithTimeout(context.Background(), time.Second*30)
+	tCtx, done := context.WithTimeout(context.Background(), time.Second*60)
 	defer done()
 
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+
 	go func() {
+		defer wg.Done()
 		assert.NoError(t, strm.Run(tCtx))
 	}()
 	require.NoError(t, inFunc(tCtx, service.NewMessage([]byte(`{"id":"hello world","content":"waddup"}`))))
 	require.NoError(t, strm.Stop(tCtx))
 
+	wg.Wait()
+
 	assert.Equal(t, `{"content":"waddup","id":"HELLO WORLD","meta":{"workflow":{"succeeded":["fooproc"]}}}`, outValue)
 	assert.Equal(t, map[string][]service.TracingEvent{
 		"barproc": {