Adds Global TTL Configuration for DynamoDB State Store

state/aws/dynamodb/dynamodb.go

@@ -48,6 +48,7 @@ type StateStore struct {
 	table            string
 	ttlAttributeName string
 	partitionKey     string
+	ttlInSeconds     *int
 
 	dynamodbClient *dynamodb.Client
 }
@@ -64,6 +65,7 @@ type dynamoDBMetadata struct {
 	Table            string `json:"table"`
 	TTLAttributeName string `json:"ttlAttributeName"`
 	PartitionKey     string `json:"partitionKey"`
+	TTLInSeconds     *int   `json:"ttlInSeconds" mapstructure:"ttlInSeconds"`
 }
 
 type putData struct {
@@ -117,6 +119,7 @@ func (d *StateStore) Init(ctx context.Context, metadata state.Metadata) error {
 	d.table = meta.Table
 	d.ttlAttributeName = meta.TTLAttributeName
 	d.partitionKey = meta.PartitionKey
+	d.ttlInSeconds = meta.TTLInSeconds
 
 	if err := d.validateTableAccess(ctx); err != nil {
 		return fmt.Errorf("error validating DynamoDB table '%s' access: %w", d.table, err)
@@ -425,6 +428,11 @@ func (d *StateStore) parseTTL(req *state.SetRequest) (*int64, error) {
 
 			return &expirationTime, nil
 		}
+		// apply global TTL if no explicit TTL in request metadata
+		if d.ttlInSeconds != nil {
+			expirationTime := time.Now().Unix() + int64(*d.ttlInSeconds)
+			return &expirationTime, nil
+		}
 	}
 
 	return nil, nil

state/aws/dynamodb/dynamodb_test.go

@@ -1206,3 +1206,170 @@ func TestMultiTx(t *testing.T) {
 		require.NoError(t, err)
 	})
 }
+
+func TestParseTTLWithDefault(t *testing.T) {
+	t.Run("Use explicit TTL from request metadata", func(t *testing.T) {
+		defaultTTL := 600
+		s := StateStore{
+			ttlAttributeName: "expiresAt",
+			ttlInSeconds:     &defaultTTL,
+		}
+
+		req := &state.SetRequest{
+			Key: "test-key",
+			Metadata: map[string]string{
+				"ttlInSeconds": "300",
+			},
+		}
+
+		ttl, err := s.parseTTL(req)
+		require.NoError(t, err)
+		require.NotNil(t, ttl)
+
+		// Should use explicit value (300), not default (600)
+		expectedTime := time.Now().Unix() + 300
+		assert.InDelta(t, expectedTime, *ttl, 2) // Allow 2 second tolerance
+	})
+
+	t.Run("Use default TTL when no explicit TTL in request", func(t *testing.T) {
+		defaultTTL := 600
+		s := StateStore{
+			ttlAttributeName: "expiresAt",
+			ttlInSeconds:     &defaultTTL,
+		}
+
+		req := &state.SetRequest{
+			Key:      "test-key",
+			Metadata: map[string]string{},
+		}
+
+		ttl, err := s.parseTTL(req)
+		require.NoError(t, err)
+		require.NotNil(t, ttl)
+
+		// Should use default value (600)
+		expectedTime := time.Now().Unix() + 600
+		assert.InDelta(t, expectedTime, *ttl, 2) // Allow 2 second tolerance
+	})
+
+	t.Run("No TTL when no default and no explicit TTL", func(t *testing.T) {
+		s := StateStore{
+			ttlAttributeName: "expiresAt",
+			ttlInSeconds:     nil, // No default configured
+		}
+
+		req := &state.SetRequest{
+			Key:      "test-key",
+			Metadata: map[string]string{},
+		}
+
+		ttl, err := s.parseTTL(req)
+		require.NoError(t, err)
+		assert.Nil(t, ttl)
+	})
+
+	t.Run("No TTL when ttlAttributeName is not set", func(t *testing.T) {
+		defaultTTL := 600
+		s := StateStore{
+			ttlAttributeName: "", // TTL not enabled in component
+			ttlInSeconds:     &defaultTTL,
+		}
+
+		req := &state.SetRequest{
+			Key: "test-key",
+			Metadata: map[string]string{
+				"ttlInSeconds": "300",
+			},
+		}
+
+		ttl, err := s.parseTTL(req)
+		require.NoError(t, err)
+		assert.Nil(t, ttl) // Should return nil when TTL not enabled
+	})
+
+	t.Run("Explicit TTL with value -1", func(t *testing.T) {
+		defaultTTL := 600
+		s := StateStore{
+			ttlAttributeName: "expiresAt",
+			ttlInSeconds:     &defaultTTL,
+		}
+
+		req := &state.SetRequest{
+			Key: "test-key",
+			Metadata: map[string]string{
+				"ttlInSeconds": "-1",
+			},
+		}
+
+		ttl, err := s.parseTTL(req)
+		require.NoError(t, err)
+		require.NotNil(t, ttl)
+
+		// -1 should result in immediate expiration (now + -1)
+		expectedTime := time.Now().Unix() - 1
+		assert.InDelta(t, expectedTime, *ttl, 2)
+	})
+
+	t.Run("Default TTL with large value", func(t *testing.T) {
+		defaultTTL := 86400 // 24 hours
+		s := StateStore{
+			ttlAttributeName: "expiresAt",
+			ttlInSeconds:     &defaultTTL,
+		}
+
+		req := &state.SetRequest{
+			Key:      "test-key",
+			Metadata: map[string]string{},
+		}
+
+		ttl, err := s.parseTTL(req)
+		require.NoError(t, err)
+		require.NotNil(t, ttl)
+
+		expectedTime := time.Now().Unix() + 86400
+		assert.InDelta(t, expectedTime, *ttl, 2)
+	})
+
+	t.Run("Error on invalid TTL value", func(t *testing.T) {
+		defaultTTL := 600
+		s := StateStore{
+			ttlAttributeName: "expiresAt",
+			ttlInSeconds:     &defaultTTL,
+		}
+
+		req := &state.SetRequest{
+			Key: "test-key",
+			Metadata: map[string]string{
+				"ttlInSeconds": "invalid",
+			},
+		}
+
+		ttl, err := s.parseTTL(req)
+		require.Error(t, err)
+		assert.Nil(t, ttl)
+		assert.Contains(t, err.Error(), "invalid syntax")
+	})
+
+	t.Run("Explicit TTL overrides default in request with empty metadata", func(t *testing.T) {
+		defaultTTL := 1200
+		s := StateStore{
+			ttlAttributeName: "expiresAt",
+			ttlInSeconds:     &defaultTTL,
+		}
+
+		req := &state.SetRequest{
+			Key: "test-key",
+			Metadata: map[string]string{
+				"ttlInSeconds": "0",
+			},
+		}
+
+		ttl, err := s.parseTTL(req)
+		require.NoError(t, err)
+		require.NotNil(t, ttl)
+
+		// Should use explicit value 0, not default
+		expectedTime := time.Now().Unix()
+		assert.InDelta(t, expectedTime, *ttl, 2)
+	})
+}

state/aws/dynamodb/metadata.yaml

@@ -36,6 +36,12 @@ metadata:
       The table attribute name which should be used for TTL.
     example: '"expiresAt"'
     type: string
+  - name: ttlInSeconds
+    required: false
+    description: |
+      Allows specifying a default Time-to-live (TTL) in seconds that will be applied to every state store request unless TTL is explicitly defined via the request metadata.
+    example: '"600"'
+    type: number
   - name: partitionKey
     required: false
     description: |