syncing up to 60a650f73b295421aa886ffdd80d00944c69e5ae

Co-authored-by: Joey Greco <[email protected]>
Co-authored-by: Carl Noel <[email protected]>

CHANGELOG.md

@@ -7,10 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## vNext
 
-- Allow branch name to be given in workflow URLs as a query param: `fetch.branch_name`
 - Upgrade Google Cloud Secret Manager Go package to `v1.14.2` (fixes "certificate_config.json: permission denied" error connecting to GCP secret manager)
 - Added support for key-pair authentication in Snowflake Plugin
 - Add support for machine-to-machine (M2M) authentication for Databricks plugin
+- Update `WaitGroup` runnable to block `Close` method on the `WaitGroup` completing (addresses `redis: client is closed` errors)
+- Allow branch name to be given in workflow HTTP requests as a header: `X-Superblocks-Branch` 
 
 ## v1.16.0
 

cmd/orchestrator/main.go

@@ -1051,7 +1051,7 @@ func main() {
 	// close server runnables, these should block while requests are still being processed
 	g.Always(grpcRunnable)
 	g.Always(httpRunnable)
-	g.Always(waitgroup.New(wg))
+	g.Always(waitgroup.New(wg, true))
 
 	// close rest of the runnables
 	g.Add(viper.GetBool("quotas.enabled"), flagsClient)

go.mod

@@ -42,7 +42,7 @@ require (
 	github.com/spf13/pflag v1.0.5
 	github.com/spf13/viper v1.16.0
 	github.com/stretchr/testify v1.9.0
-	github.com/superblocksteam/run v0.0.6
+	github.com/superblocksteam/run v0.0.7
 	github.com/thejerf/abtime v1.0.7
 	github.com/titanous/json5 v1.0.0
 	go.kuoruan.net/v8go-polyfills v0.5.0

go.sum

@@ -638,8 +638,8 @@ github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8
 github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
 github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8=
 github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU=
-github.com/superblocksteam/run v0.0.6 h1:ZA+MAxRhAmGrtsS80ehXBXr3GleHy16QIt9ytwLickw=
-github.com/superblocksteam/run v0.0.6/go.mod h1:vJnCTxFlZIT+5k8062ABRdLTJxRcahcv7RYCe+wSGLU=
+github.com/superblocksteam/run v0.0.7 h1:D9z31GpM1dGeeLcHUSWP9Lv0ZsiOQYIehorMolYrLu8=
+github.com/superblocksteam/run v0.0.7/go.mod h1:vJnCTxFlZIT+5k8062ABRdLTJxRcahcv7RYCe+wSGLU=
 github.com/testcontainers/testcontainers-go v0.31.0 h1:W0VwIhcEVhRflwL9as3dhY6jXjVCA27AkmbnZ+UTh3U=
 github.com/testcontainers/testcontainers-go v0.31.0/go.mod h1:D2lAoA0zUFiSY+eAflqK5mcUx/A5hrrORaEQrd0SefI=
 github.com/testcontainers/testcontainers-go/modules/compose v0.31.0 h1:H74o3HisnApIDQx7sWibGzOl/Oo0By8DjyVeUf3qd6I=

internal/transport/http.go

@@ -21,7 +21,6 @@ import (
 const (
 	TransformedTokenQueryParam                 = "fetch.token"
 	TransformedTestQueryParam                  = "fetch.test"
-	TransformedBranchNameQueryParam            = "fetch.branch_name"
 	TransformedEnvironmentQueryParam           = "fetch.profile.environment"
 	TransformedProfileNameQueryParam           = "fetch.profile.name"
 	TransformedProfileIdQueryParam             = "fetch.profile.id"
@@ -34,7 +33,6 @@ var (
 	knownQueryParams = map[string]bool{
 		TransformedTokenQueryParam:                 true,
 		TransformedTestQueryParam:                  true,
-		TransformedBranchNameQueryParam:            true,
 		TransformedEnvironmentQueryParam:           true,
 		TransformedOptionsAsync:                    true,
 		TransformedProfileNameQueryParam:           true,
@@ -262,10 +260,6 @@ func transformWorkflowRequest(r *http.Request, version string) (err error) {
 		request.Request.(*apiv1.ExecuteRequest_Fetch_).Fetch.Profile.Id = &profileID
 	}
 
-	if branchName := r.URL.Query().Get(TransformedBranchNameQueryParam); branchName != "" {
-		request.Request.(*apiv1.ExecuteRequest_Fetch_).Fetch.BranchName = &branchName
-	}
-
 	if include := r.URL.Query().Get(TransformedOptionsIncludeEventOutputsParam); include != "" {
 		includeBool, err := strconv.ParseBool(include)
 		if err != nil {
@@ -290,6 +284,11 @@ func transformWorkflowRequest(r *http.Request, version string) (err error) {
 		request.Options.Async = asyncBool
 	}
 
+	// headers
+	if branchName := r.Header.Get("X-Superblocks-Branch"); branchName != "" {
+		request.Request.(*apiv1.ExecuteRequest_Fetch_).Fetch.BranchName = &branchName
+	}
+
 	var buf bytes.Buffer
 	{
 		if err := marshaler.Marshal(&buf, request); err != nil {

internal/transport/http_test.go

@@ -421,10 +421,13 @@ func TestTransformWorkflowRequest(t *testing.T) {
 			},
 		},
 		{
-			name: "with query param fetch.branch_name",
+			name: "with header X-Superblocks-Branch",
 			httpRequest: &http.Request{
 				Method: http.MethodPost,
-				URL:    &url.URL{Path: "/foo/bar", RawQuery: "fetch.branch_name=foo"},
+				URL:    &url.URL{Path: "/foo/bar"},
+				Header: http.Header{
+					"X-Superblocks-Branch": []string{"foo"},
+				},
 			},
 			version: "v2",
 			expectedExecuteRequest: &apiv1.ExecuteRequest{
@@ -446,7 +449,7 @@ func TestTransformWorkflowRequest(t *testing.T) {
 			name: "with multiple query params",
 			httpRequest: &http.Request{
 				Method: http.MethodPost,
-				URL:    &url.URL{Path: "/foo/bar", RawQuery: "fetch.profile.environment=foo&fetch.branch_name=bar"},
+				URL:    &url.URL{Path: "/foo/bar", RawQuery: "fetch.profile.environment=foo&fetch.profile.id=bar"},
 			},
 			version: "v2",
 			expectedExecuteRequest: &apiv1.ExecuteRequest{
@@ -460,8 +463,8 @@ func TestTransformWorkflowRequest(t *testing.T) {
 						ViewMode: apiv1.ViewMode_VIEW_MODE_DEPLOYED,
 						Profile: &commonv1.Profile{
 							Environment: proto.String("foo"),
+							Id:          proto.String("bar"),
 						},
-						BranchName: proto.String("bar"),
 					},
 				},
 			},

postman/collection.json

@@ -1,6 +1,6 @@
 {
 	"info": {
-		"_postman_id": "328e2cc0-8a1c-469a-9654-ced153435781",
+		"_postman_id": "2b1c3ae9-cc25-4afb-981d-61463840093d",
 		"name": "collection.json - [HTTP] Superblocks Orchestrator",
 		"schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json",
 		"_exporter_id": "25579613"
@@ -1539,7 +1539,13 @@
 									"type": "noauth"
 								},
 								"method": "POST",
-								"header": [],
+								"header": [
+									{
+										"key": "X-Superblocks-Branch",
+										"value": "feature-branch-1",
+										"type": "text"
+									}
+								],
 								"body": {
 									"mode": "raw",
 									"raw": "{}\n",
@@ -1550,7 +1556,7 @@
 									}
 								},
 								"url": {
-									"raw": "{{orchestrator_scheme}}://{{orchestrator_host}}:{{orchestrator_port}}/v1/workflows/00000000-0000-0000-0000-000000000004?sb-auth=token&profile=known&multiple=one&multiple=1&queryParamOne=OVERRIDE&fetch.branch_name=feature-branch-1",
+									"raw": "{{orchestrator_scheme}}://{{orchestrator_host}}:{{orchestrator_port}}/v1/workflows/00000000-0000-0000-0000-000000000004?sb-auth=token&profile=known&multiple=one&multiple=1&queryParamOne=OVERRIDE",
 									"protocol": "{{orchestrator_scheme}}",
 									"host": [
 										"{{orchestrator_host}}"
@@ -1581,10 +1587,6 @@
 										{
 											"key": "queryParamOne",
 											"value": "OVERRIDE"
-										},
-										{
-											"key": "fetch.branch_name",
-											"value": "feature-branch-1"
 										}
 									]
 								}
@@ -2262,7 +2264,13 @@
 							"type": "noauth"
 						},
 						"method": "POST",
-						"header": [],
+						"header": [
+							{
+								"key": "X-Superblocks-Branch",
+								"value": "feature-branch-1",
+								"type": "text"
+							}
+						],
 						"body": {
 							"mode": "raw",
 							"raw": "{}\n",
@@ -2273,7 +2281,7 @@
 							}
 						},
 						"url": {
-							"raw": "{{orchestrator_scheme}}://{{orchestrator_host}}:{{orchestrator_port}}/v2/execute/00000000-0000-0000-0000-000000000004?sb-auth=token&profile=known&multiple=one&multiple=1&queryParamOne=OVERRIDE&fetch.branch_name=feature-branch-1",
+							"raw": "{{orchestrator_scheme}}://{{orchestrator_host}}:{{orchestrator_port}}/v2/execute/00000000-0000-0000-0000-000000000004?sb-auth=token&profile=known&multiple=one&multiple=1&queryParamOne=OVERRIDE",
 							"protocol": "{{orchestrator_scheme}}",
 							"host": [
 								"{{orchestrator_host}}"
@@ -2304,10 +2312,6 @@
 								{
 									"key": "queryParamOne",
 									"value": "OVERRIDE"
-								},
-								{
-									"key": "fetch.branch_name",
-									"value": "feature-branch-1"
 								}
 							]
 						}

workers/golang/go.mod

@@ -47,7 +47,7 @@ require (
 	github.com/spf13/viper v1.16.0
 	github.com/stretchr/testify v1.9.0
 	github.com/superblocksteam/agent v0.0.0-00010101000000-000000000000
-	github.com/superblocksteam/run v0.0.6
+	github.com/superblocksteam/run v0.0.7
 	go.opentelemetry.io/otel v1.29.0
 	go.opentelemetry.io/otel/sdk v1.29.0
 	go.opentelemetry.io/otel/trace v1.29.0

workers/golang/go.sum

@@ -167,8 +167,8 @@ github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsT
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8=
 github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU=
-github.com/superblocksteam/run v0.0.6 h1:ZA+MAxRhAmGrtsS80ehXBXr3GleHy16QIt9ytwLickw=
-github.com/superblocksteam/run v0.0.6/go.mod h1:vJnCTxFlZIT+5k8062ABRdLTJxRcahcv7RYCe+wSGLU=
+github.com/superblocksteam/run v0.0.7 h1:D9z31GpM1dGeeLcHUSWP9Lv0ZsiOQYIehorMolYrLu8=
+github.com/superblocksteam/run v0.0.7/go.mod h1:vJnCTxFlZIT+5k8062ABRdLTJxRcahcv7RYCe+wSGLU=
 github.com/tklauser/go-sysconf v0.3.12 h1:0QaGUFOdQaIVdPgfITYzaTegZvdCjmYO52cSFAEVmqU=
 github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI=
 github.com/tklauser/numcpus v0.6.1 h1:ng9scYS7az0Bk4OZLvrNXNSAO2Pxr1XXRAPyjhIx+Fk=