Add Saga Handler (#192)

The saga pattern descibes a saga -- a series of steps -- that may be
required for an API request.  It's well aware that some intermediate
step can go wrong, most likely quotas, and prior steps will need to be
compensated for to avoid resource leaks etc.  This implements that in a
nice generic way so it can be used by all complex handlers that do lots
of things at once.

Author: spjmurray

pkg/server/saga/saga.go

@@ -0,0 +1,93 @@
+/*
+Copyright 2025 the Unikorn 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 saga
+
+import (
+	"context"
+
+	"sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+// ActionFunc is a generic action/compensation function.
+// They will typically be bound receivers so that saga steps can
+// share state between themselves.
+type ActionFunc func(ctx context.Context) error
+
+// Action is a single step in a saga.
+type Action struct {
+	// name is used for logging so we can see what went wrong.
+	name string
+	// action is what is executed on the good path.
+	action ActionFunc
+	// compensate is run if a subsequent action in the saga fails
+	// and can undo any state changes that need to be rewound.
+	// May be nil.
+	compensate ActionFunc
+}
+
+// NewAction creates a new action.
+func NewAction(name string, action, compensate ActionFunc) Action {
+	return Action{
+		name:       name,
+		action:     action,
+		compensate: compensate,
+	}
+}
+
+// Handler implements a saga, a set of steps to achieve a desired outcome
+// and a set of steps to undo any state changes on failure of an action.
+type Handler interface {
+	Actions() []Action
+}
+
+// Run implements the saga algorithm.
+func Run(ctx context.Context, handler Handler) error {
+	log := log.FromContext(ctx)
+
+	actions := handler.Actions()
+
+	// Do each action in order...
+	for i := range actions {
+		if err := actions[i].action(ctx); err != nil {
+			// If something went wrong we need to undo all prior steps
+			// to compensate for any changed state e.g. quota allocations.
+			for j := i - 1; j >= 0; j-- {
+				if actions[j].compensate == nil {
+					continue
+				}
+
+				if cerr := actions[j].compensate(ctx); cerr != nil {
+					// You see this in your logs, you're going to have to
+					// do some manual unpicking!
+					// TODO: we could add a retry in here for transient errors
+					// (and the actual action itself), but we aware the client
+					// and server will have a response timeout, so perhaps
+					// adding the compensation action to a log for aysnchronous
+					// handling may be better in future.
+					log.Error(cerr, "compensating action failed", "name", actions[j].name)
+					return err
+				}
+			}
+
+			// Always return the error that caused failure, which will most likely
+			// be something useful to user like quota allocation failures.
+			return err
+		}
+	}
+
+	return nil
+}

pkg/server/saga/saga_test.go

@@ -0,0 +1,145 @@
+/*
+Copyright 2025 the Unikorn 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 saga_test
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/unikorn-cloud/core/pkg/server/saga"
+)
+
+var (
+	errFailAction     = errors.New("fail action")
+	errFailCompensate = errors.New("fail compensation")
+)
+
+type Handler struct {
+	action1Result error
+	action2Result error
+	action3Result error
+
+	compensate1Result error
+	compensate2Result error
+
+	action1Called bool
+	action2Called bool
+	action3Called bool
+
+	compensate1Called bool
+	compensate2Called bool
+}
+
+func (h *Handler) action1(ctx context.Context) error {
+	h.action1Called = true
+	return h.action1Result
+}
+
+func (h *Handler) action2(ctx context.Context) error {
+	h.action2Called = true
+	return h.action2Result
+}
+
+func (h *Handler) action3(ctx context.Context) error {
+	h.action3Called = true
+	return h.action3Result
+}
+
+func (h *Handler) compensate1(ctx context.Context) error {
+	h.compensate1Called = true
+	return h.compensate1Result
+}
+
+func (h *Handler) compensate2(ctx context.Context) error {
+	h.compensate2Called = true
+	return h.compensate2Result
+}
+
+func (h *Handler) Actions() []saga.Action {
+	return []saga.Action{
+		saga.NewAction("action1", h.action1, h.compensate1),
+		saga.NewAction("action2", h.action2, h.compensate2),
+		saga.NewAction("action3", h.action3, nil),
+	}
+}
+
+// TestSaga ensures all actions are called.
+func TestSaga(t *testing.T) {
+	t.Parallel()
+
+	h := &Handler{}
+
+	require.NoError(t, saga.Run(t.Context(), h))
+	require.True(t, h.action1Called)
+	require.True(t, h.action2Called)
+	require.True(t, h.action3Called)
+	require.False(t, h.compensate1Called)
+	require.False(t, h.compensate2Called)
+}
+
+// TestSagaFailAction1 ensures compensating actions are correctly run.
+func TestSagaFailAction1(t *testing.T) {
+	t.Parallel()
+
+	h := &Handler{
+		action1Result: errFailAction,
+	}
+
+	require.ErrorIs(t, saga.Run(t.Context(), h), errFailAction)
+	require.True(t, h.action1Called)
+	require.False(t, h.action2Called)
+	require.False(t, h.action3Called)
+	require.False(t, h.compensate1Called)
+	require.False(t, h.compensate2Called)
+}
+
+// TestSagaFailAction3 ensures compensating actions are correctly run.
+func TestSagaFailAction3(t *testing.T) {
+	t.Parallel()
+
+	h := &Handler{
+		action3Result: errFailAction,
+	}
+
+	require.ErrorIs(t, saga.Run(t.Context(), h), errFailAction)
+	require.True(t, h.action1Called)
+	require.True(t, h.action2Called)
+	require.True(t, h.action3Called)
+	require.True(t, h.compensate1Called)
+	require.True(t, h.compensate2Called)
+}
+
+// TestSagaFailCompensation2 tests that compensation errors short circuit
+// and that the error returned is that of the failing action.
+func TestSagaFailCompensation2(t *testing.T) {
+	t.Parallel()
+
+	h := &Handler{
+		action3Result:     errFailAction,
+		compensate2Result: errFailCompensate,
+	}
+
+	require.ErrorIs(t, saga.Run(t.Context(), h), errFailAction)
+	require.True(t, h.action1Called)
+	require.True(t, h.action2Called)
+	require.True(t, h.action3Called)
+	require.False(t, h.compensate1Called)
+	require.True(t, h.compensate2Called)
+}