From 86fa588e5782e67dcaaa13367a08fe28745751cf Mon Sep 17 00:00:00 2001 From: CMGS Date: Thu, 10 Sep 2026 01:17:17 +0900 Subject: [PATCH] test: pin the WAL processing replay and the AddNode cache eviction ProcessingCreatedHandler is replayed against an embedded etcd: only the abandoned processing record of the recovering instance is released, a live request of another instance keeps its count and its journal, and a second recovery has nothing left. AddNode drops the cached engine when the resource plugins refuse the node, so a retry dials again instead of reusing the stale client. --- cluster/calcium/node_test.go | 31 ++++++++++++ cluster/calcium/wal_test.go | 98 ++++++++++++++++++++++++++++++++++++ 2 files changed, 129 insertions(+) diff --git a/cluster/calcium/node_test.go b/cluster/calcium/node_test.go index a1980e09..13b8f34b 100644 --- a/cluster/calcium/node_test.go +++ b/cluster/calcium/node_test.go @@ -93,6 +93,37 @@ func TestAddNodeRedoesTheStalePluginMetadata(t *testing.T) { store.AssertExpectations(t) } +func TestAddNodeEvictsTheCachedEngineOnFailure(t *testing.T) { + c := NewTestCluster() + c.config.ConnectionTimeout = time.Second + ctx := t.Context() + factory.InitEngineCache(ctx, c.config, nil) + + nodename := "nodename" + opts := &types.AddNodeOptions{ + Nodename: nodename, + Podname: "podname", + Endpoint: fmt.Sprintf("mock://%s", nodename), + } + + rmgr := c.rmgr.(*resourcemocks.Manager) + rmgr.On("AddNode", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil, types.ErrMockError).Once() + _, err := c.AddNode(ctx, opts) + assert.Error(t, err) + assert.Nil(t, factory.GetEngineFromCache(ctx, opts.Endpoint), "a failed AddNode must not leave the endpoint cached") + + rmgr.On("AddNode", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(resourcetypes.Resources{}, nil).Once() + rmgr.On("GetNodesMetrics", mock.Anything, mock.Anything).Return(nil, nil).Maybe() + store := c.store.(*storemocks.Store) + store.On("AddNode", mock.Anything, mock.Anything).Return(&types.Node{NodeMeta: types.NodeMeta{Name: nodename}}, nil).Once() + + _, err = c.AddNode(ctx, opts) + assert.NoError(t, err) + assert.NotNil(t, factory.GetEngineFromCache(ctx, opts.Endpoint), "an added node keeps its engine cached") + rmgr.AssertExpectations(t) + store.AssertExpectations(t) +} + func TestRemoveNode(t *testing.T) { c := NewTestCluster() ctx := t.Context() diff --git a/cluster/calcium/wal_test.go b/cluster/calcium/wal_test.go index 3a775c3c..778d24c8 100644 --- a/cluster/calcium/wal_test.go +++ b/cluster/calcium/wal_test.go @@ -2,25 +2,37 @@ package calcium import ( "context" + "fmt" "maps" + "path/filepath" "strings" "sync" "testing" "testing/synctest" + "time" "github.com/cockroachdb/errors" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" enginemocks "github.com/projecteru2/core/engine/mocks" enginetypes "github.com/projecteru2/core/engine/types" lockmocks "github.com/projecteru2/core/lock/mocks" + "github.com/projecteru2/core/log" resourcemocks "github.com/projecteru2/core/resource/mocks" resourcetypes "github.com/projecteru2/core/resource/types" + "github.com/projecteru2/core/store" + storecommon "github.com/projecteru2/core/store/common" + "github.com/projecteru2/core/store/etcdv3/embedded" + storefactory "github.com/projecteru2/core/store/factory" storemocks "github.com/projecteru2/core/store/mocks" "github.com/projecteru2/core/types" + "github.com/projecteru2/core/utils" ) +const addressPrefix = "/wal/%s/" + func TestHandleWorkloadResourceAllocatedMultipleNodes(t *testing.T) { c := NewTestCluster() store := c.store.(*storemocks.Store) @@ -413,6 +425,53 @@ func TestHandleCreateLambdaKeepsEntryUntilRemoved(t *testing.T) { }) } +func TestProcessingCreatedHandlerDropsTheAbandonedRecordOnly(t *testing.T) { + const appname, entryname, nodename = "app", "entry", "node" + + ctx := t.Context() + config := types.Config{ + GlobalTimeout: 30 * time.Second, + MaxConcurrency: 100, + Etcd: types.EtcdConfig{Prefix: "/eru-test", LockPrefix: "/eru-test-lock"}, + } + stor := newEmbeddedStore(t, config) + recovering := newJournaledCluster(t, config, stor, "10.22.12.34:5001") + peer := newJournaledCluster(t, config, stor, "10.22.12.35:5001") + logger := log.WithFunc("calcium.wal_test") + + abandoned := newRequestProcessing(appname, entryname, nodename) + _, err := recovering.journal(ctx, logger, eventProcessingCreated, abandoned) + require.NoError(t, err) + require.NoError(t, stor.CreateProcessing(ctx, abandoned, 3)) + + finished := newRequestProcessing(appname, entryname, nodename) + commit, err := recovering.journal(ctx, logger, eventProcessingCreated, finished) + require.NoError(t, err) + require.NoError(t, stor.CreateProcessing(ctx, finished, 2)) + require.NoError(t, stor.DeleteProcessing(ctx, finished)) + commit() + + inflight := newRequestProcessing(appname, entryname, nodename) + _, err = peer.journal(ctx, logger, eventProcessingCreated, inflight) + require.NoError(t, err) + require.NoError(t, stor.CreateProcessing(ctx, inflight, 5)) + + require.Len(t, journalEntries(t, stor, recovering.serviceAddress), 1, "the finished request commits its entry away, the crashed one leaves it behind") + require.Equal(t, map[string]int{nodename: 8}, deployStatus(t, stor, appname, entryname), "the crashed request still counts against the node") + + recovering.DisasterRecover(ctx) + + keys, err := stor.ListPrefix(ctx, filepath.Join(storecommon.WorkloadProcessingPrefix, appname, entryname)+"/") + require.NoError(t, err) + assert.Equal(t, []string{storecommon.ProcessingKey(inflight)}, keys, "only the abandoned ident is released") + assert.Equal(t, map[string]int{nodename: 5}, deployStatus(t, stor, appname, entryname), "the live request keeps its count") + assert.Empty(t, journalEntries(t, stor, recovering.serviceAddress), "a replayed entry is dropped") + assert.Len(t, journalEntries(t, stor, peer.serviceAddress), 1, "another instance's journal is not replayed") + + recovering.DisasterRecover(ctx) + assert.Equal(t, map[string]int{nodename: 5}, deployStatus(t, stor, appname, entryname), "a second recovery has nothing left to replay") +} + func enableTestWAL(t *testing.T, c *Calcium) { mockWALStore(c.store.(*storemocks.Store)) journal, err := enableWAL(t.Context(), c.config, c, c.store) @@ -461,3 +520,42 @@ func mockWALStore(store *storemocks.Store) { return keys, nil }).Maybe() } + +func newEmbeddedStore(t *testing.T, config types.Config) store.Store { + etcd, err := embedded.New(t.TempDir()) + require.NoError(t, err) + t.Cleanup(etcd.Close) + + stor, err := storefactory.NewStore(t.Context(), config, etcd) + require.NoError(t, err) + return stor +} + +func newJournaledCluster(t *testing.T, config types.Config, stor store.Store, address string) *Calcium { + c := &Calcium{config: config, store: stor, serviceAddress: address} + journal, err := enableWAL(t.Context(), config, c, stor) + require.NoError(t, err) + c.wal = journal + return c +} + +func newRequestProcessing(appname, entryname, nodename string) *types.Processing { + opts := &types.DeployOptions{ + Name: appname, + Entrypoint: &types.Entrypoint{Name: entryname}, + ProcessIdent: utils.RandomString(16), + } + return opts.GetProcessing(nodename) +} + +func journalEntries(t *testing.T, stor store.Store, address string) []string { + keys, err := stor.ListPrefix(t.Context(), fmt.Sprintf(addressPrefix, address)) + require.NoError(t, err) + return keys +} + +func deployStatus(t *testing.T, stor store.Store, appname, entryname string) map[string]int { + status, err := stor.GetDeployStatus(t.Context(), appname, entryname) + require.NoError(t, err) + return status +}