feat(config): add configurable expiration settings for workflow runs

Signed-off-by: Jordan McClintock <jordan@defenseunicorns.com>
Assisted-by: OpenAI Codex

Author: corang

app/controlplane/cmd/main.go

@@ -158,9 +158,7 @@ func main() {
 	}
 	defer cleanup()
 
-	// Run an expiration job every minute that expires unfinished runs older than 1 hour
-	// TODO: Make it configurable from the application config
-	app.runsExpirer.Run(ctx, &biz.WorkflowRunExpirerOpts{CheckInterval: 1 * time.Minute, ExpirationWindow: 1 * time.Hour})
+	app.runsExpirer.Run(ctx, workflowRunExpirerOpts(&bc))
 
 	// Sync user access
 	go func() {
@@ -209,6 +207,18 @@ func toPtr[T any](v T) *T {
 	return &v
 }
 
+func workflowRunExpirerOpts(c *conf.Bootstrap) *biz.WorkflowRunExpirerOpts {
+	opts := &biz.WorkflowRunExpirerOpts{}
+	if configured := c.GetAttestations().GetWorkflowRunExpirationWindow(); configured != nil {
+		opts.ExpirationWindow = configured.AsDuration()
+	}
+	if configured := c.GetAttestations().GetWorkflowRunExpirationCheckInterval(); configured != nil {
+		opts.CheckInterval = configured.AsDuration()
+	}
+
+	return opts
+}
+
 type app struct {
 	*kratos.App
 	// Periodic job that expires unfinished attestation processes older than a given threshold

app/controlplane/cmd/main_test.go

@@ -0,0 +1,125 @@
+// Copyright 2024-2026 The Chainloop Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	conf "github.com/chainloop-dev/chainloop/app/controlplane/internal/conf/controlplane/config/v1"
+	"github.com/go-kratos/kratos/v2/config"
+	"github.com/go-kratos/kratos/v2/config/file"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/types/known/durationpb"
+)
+
+func TestWorkflowRunExpirerOptsMapping(t *testing.T) {
+	tests := []struct {
+		name              string
+		config            *conf.Bootstrap
+		want              time.Duration
+		wantCheckInterval time.Duration
+	}{
+		{
+			name:   "missing attestations configuration",
+			config: &conf.Bootstrap{},
+			want:   0,
+		},
+		{
+			name: "unset expiration window",
+			config: &conf.Bootstrap{
+				Attestations: &conf.Attestations{},
+			},
+			want: 0,
+		},
+		{
+			name: "configured expiration window",
+			config: &conf.Bootstrap{
+				Attestations: &conf.Attestations{
+					WorkflowRunExpirationWindow:        durationpb.New(2 * time.Hour),
+					WorkflowRunExpirationCheckInterval: durationpb.New(30 * time.Second),
+				},
+			},
+			wantCheckInterval: 30 * time.Second,
+			want:              2 * time.Hour,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			opts := workflowRunExpirerOpts(tt.config)
+			assert.Equal(t, tt.want, opts.ExpirationWindow)
+			assert.Equal(t, tt.wantCheckInterval, opts.CheckInterval)
+		})
+	}
+}
+
+func TestWorkflowRunExpirationDurationValidation(t *testing.T) {
+	validator, err := newProtoValidator()
+	require.NoError(t, err)
+
+	for _, duration := range []time.Duration{0, -time.Second} {
+		t.Run(duration.String(), func(t *testing.T) {
+			bootstrap := &conf.Bootstrap{
+				Attestations: &conf.Attestations{
+					WorkflowRunExpirationWindow: durationpb.New(duration),
+				},
+			}
+			require.Error(t, validator.Validate(bootstrap))
+
+			bootstrap = &conf.Bootstrap{
+				Attestations: &conf.Attestations{
+					WorkflowRunExpirationCheckInterval: durationpb.New(duration),
+				},
+			}
+			require.Error(t, validator.Validate(bootstrap))
+		})
+	}
+}
+
+func TestMalformedWorkflowRunExpirationWindow(t *testing.T) {
+	configPath := filepath.Join(t.TempDir(), "config.yaml")
+	require.NoError(t, os.WriteFile(configPath, []byte(
+		"attestations:\n  workflow_run_expiration_window: definitely-not-a-duration\n",
+	), 0o600))
+
+	c := config.New(config.WithSource(file.NewSource(configPath)))
+	t.Cleanup(func() {
+		require.NoError(t, c.Close())
+	})
+	require.NoError(t, c.Load())
+
+	var bootstrap conf.Bootstrap
+	require.Error(t, c.Scan(&bootstrap))
+}
+
+func TestMalformedWorkflowRunExpirationCheckInterval(t *testing.T) {
+	configPath := filepath.Join(t.TempDir(), "config.yaml")
+	require.NoError(t, os.WriteFile(configPath, []byte(
+		"attestations:\n  workflow_run_expiration_check_interval: definitely-not-a-duration\n",
+	), 0o600))
+
+	c := config.New(config.WithSource(file.NewSource(configPath)))
+	t.Cleanup(func() {
+		require.NoError(t, c.Close())
+	})
+	require.NoError(t, c.Load())
+
+	var bootstrap conf.Bootstrap
+	require.Error(t, c.Scan(&bootstrap))
+}

app/controlplane/configs/config.devel.yaml

@@ -125,3 +125,7 @@ ui_dashboard_url: http://localhost:3000
 
 attestations:
   skip_db_storage: true
+  # Unfinished workflow runs older than this are marked as expired.
+  workflow_run_expiration_window: 1h
+  # How often unfinished workflow runs are checked for expiration.
+  workflow_run_expiration_check_interval: 1m

app/controlplane/configs/samples/config.yaml

@@ -98,4 +98,10 @@ enable_profiler: true
 #   # When true, skip writing the attestation bundle to the database; the bundle is
 #   # stored exclusively in the configured CAS backend. Has no effect when the
 #   # workflow run's CAS backend is inline.
-#   skip_db_storage: false
+#   skip_db_storage: false
+#   # Maximum age of an unfinished workflow run before it is marked as expired.
+#   # Defaults to 1 hour when omitted.
+#   # Interval between expiration checks for unfinished workflow runs.
+#   # Defaults to 1 minute when omitted.
+#   workflow_run_expiration_check_interval: 1m
+#   workflow_run_expiration_window: 1h

app/controlplane/internal/conf/controlplane/config/v1/conf.pb.go

@@ -160,6 +160,16 @@ message Attestations {
   // from exhausting the control plane's memory. Values <= 0 select the
   // built-in default.
   int64 policy_evaluations_max_inline_bytes = 2;
+
+  // Maximum amount of time an unfinished workflow run can remain active before
+  // it is marked as expired. When unset, the control plane defaults to 1 hour.
+  google.protobuf.Duration workflow_run_expiration_window = 3
+      [(buf.validate.field).duration.gt = {}];
+
+  // Interval between checks for unfinished workflow runs. When unset, the
+  // control plane defaults to 1 minute.
+  google.protobuf.Duration workflow_run_expiration_check_interval = 4
+      [(buf.validate.field).duration.gt = {}];
 }
 
 message OperationAuthorizationProvider {

app/controlplane/internal/conf/controlplane/config/v1/conf.proto

@@ -180,18 +180,43 @@ type WorkflowRunExpirerUseCase struct {
 	logger         *log.Helper
 }
 
+const (
+	defaultWorkflowRunExpirationWindow        = time.Hour
+	defaultWorkflowRunExpirationCheckInterval = time.Minute
+)
+
 type WorkflowRunExpirerOpts struct {
-	// Maximum time threshold for what a workflowRun will be considered expired
+	// Maximum time threshold for what a workflowRun will be considered expired. Defaults to 1 hour.
 	ExpirationWindow time.Duration
-	CheckInterval    time.Duration
+	// Interval between expiration sweeps. Defaults to 1 minute.
+	CheckInterval time.Duration
 }
 
 func NewWorkflowRunExpirerUseCase(wfrRepo WorkflowRunRepo, po PromObservable, logger log.Logger) *WorkflowRunExpirerUseCase {
 	logger = log.With(logger, "component", "biz.WorkflowRunExpirer")
 	return &WorkflowRunExpirerUseCase{wfrRepo, po, log.NewHelper(logger)}
 }
 
+func resolveWorkflowRunExpirerOpts(opts *WorkflowRunExpirerOpts) WorkflowRunExpirerOpts {
+	resolved := WorkflowRunExpirerOpts{
+		ExpirationWindow: defaultWorkflowRunExpirationWindow,
+		CheckInterval:    defaultWorkflowRunExpirationCheckInterval,
+	}
+	if opts == nil {
+		return resolved
+	}
+	if opts.ExpirationWindow > 0 {
+		resolved.ExpirationWindow = opts.ExpirationWindow
+	}
+	if opts.CheckInterval > 0 {
+		resolved.CheckInterval = opts.CheckInterval
+	}
+
+	return resolved
+}
+
 func (uc *WorkflowRunExpirerUseCase) Run(ctx context.Context, opts *WorkflowRunExpirerOpts) {
+	resolvedOpts := resolveWorkflowRunExpirerOpts(opts)
 	timer := time.NewTimer(0)
 
 	go func() {
@@ -201,18 +226,18 @@ func (uc *WorkflowRunExpirerUseCase) Run(ctx context.Context, opts *WorkflowRunE
 			case <-ctx.Done():
 				return
 			case <-timer.C:
-				threshold := time.Now().Add(-opts.ExpirationWindow)
+				threshold := time.Now().Add(-resolvedOpts.ExpirationWindow)
 
 				if err := uc.ExpirationSweep(ctx, threshold); err != nil {
 					uc.logger.Error(err)
 				}
 			}
 
-			timer.Reset(opts.CheckInterval)
+			timer.Reset(resolvedOpts.CheckInterval)
 		}
 	}()
 
-	uc.logger.Infof("periodic check enabled. interval=%s, expirationWindow=%s", opts.CheckInterval, opts.ExpirationWindow)
+	uc.logger.Infof("periodic check enabled. interval=%s, expirationWindow=%s", resolvedOpts.CheckInterval, resolvedOpts.ExpirationWindow)
 }
 
 // ExpirationSweep looks for runs older than the provider time and marks them as expired