diff --git a/rest-api/flow/internal/nicoapi/grpc.go b/rest-api/flow/internal/nicoapi/grpc.go index 160fdceebe..0cbac07ca3 100644 --- a/rest-api/flow/internal/nicoapi/grpc.go +++ b/rest-api/flow/internal/nicoapi/grpc.go @@ -8,6 +8,8 @@ import ( "errors" "fmt" "os" + "slices" + "strings" "sync" "testing" "time" @@ -19,6 +21,7 @@ import ( "github.com/rs/zerolog/log" "google.golang.org/grpc" "google.golang.org/grpc/credentials" + "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/emptypb" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -54,6 +57,69 @@ type grpcClient struct { grpcTimeout time.Duration } +// batchingForgeClient keeps limit handling below the Flow client methods so +// direct ForgeClient calls use the same batching behavior as the convenience +// methods in this package. +type batchingForgeClient struct { + corev1.ForgeClient + + maxFindByIDsMu sync.Mutex + maxFindByIDs uint32 + maxFindByIDsLoaded bool +} + +func newBatchingForgeClient(client corev1.ForgeClient) corev1.ForgeClient { + return &batchingForgeClient{ForgeClient: client} +} + +func (c *batchingForgeClient) FindMachinesByIds( + ctx context.Context, + request *corev1.MachinesByIdsRequest, + options ...grpc.CallOption, +) (*corev1.MachineList, error) { + machineIDs := protoIDsToStrings(request.GetMachineIds()) + machines, err := findByIDBatches(ctx, c.loadMaxFindByIDs, machineIDs, + func(ctx context.Context, batch []string) ([]*corev1.Machine, error) { + return c.fetchMachinesByIDs(ctx, request, batch, options...) + }) + if err != nil { + return nil, err + } + return &corev1.MachineList{Machines: machines}, nil +} + +func (c *batchingForgeClient) FindSwitchesByIds( + ctx context.Context, + request *corev1.SwitchesByIdsRequest, + options ...grpc.CallOption, +) (*corev1.SwitchList, error) { + switchIDs := protoIDsToStrings(request.GetSwitchIds()) + switches, err := findByIDBatches(ctx, c.loadMaxFindByIDs, switchIDs, + func(ctx context.Context, batch []string) ([]*corev1.Switch, error) { + return c.fetchSwitchesByIDs(ctx, request, batch, options...) + }) + if err != nil { + return nil, err + } + return &corev1.SwitchList{Switches: switches}, nil +} + +func (c *batchingForgeClient) FindPowerShelvesByIds( + ctx context.Context, + request *corev1.PowerShelvesByIdsRequest, + options ...grpc.CallOption, +) (*corev1.PowerShelfList, error) { + shelfIDs := protoIDsToStrings(request.GetPowerShelfIds()) + shelves, err := findByIDBatches(ctx, c.loadMaxFindByIDs, shelfIDs, + func(ctx context.Context, batch []string) ([]*corev1.PowerShelf, error) { + return c.fetchPowerShelvesByIDs(ctx, request, batch, options...) + }) + if err != nil { + return nil, err + } + return &corev1.PowerShelfList{PowerShelves: shelves}, nil +} + var testingMsgOnce sync.Once // NewClient creates a GRPC connection pool to nico-core-api. Returning success does not mean that we have yet made an actual connection; @@ -88,7 +154,10 @@ func NewClient(grpcTimeout time.Duration) (Client, error) { return nil, fmt.Errorf("Unable to connect to nico-core-api: %w", err) } - return &grpcClient{gclient: corev1.NewForgeClient(conn), grpcTimeout: grpcTimeout}, nil + return &grpcClient{ + gclient: newBatchingForgeClient(corev1.NewForgeClient(conn)), + grpcTimeout: grpcTimeout, + }, nil } // GetMachines retrieves all machines known by nico-core-api @@ -185,6 +254,120 @@ func (c *grpcClient) Version(ctx context.Context) (string, error) { return res.GetBuildVersion(), nil } +// loadMaxFindByIDs lazily loads and caches Core's effective request limit. +// Failed loads are not cached, so a transient Version failure can recover on a +// later lookup. The mutex also coalesces concurrent first loads into one RPC. +func (c *batchingForgeClient) loadMaxFindByIDs(ctx context.Context) (uint32, error) { + c.maxFindByIDsMu.Lock() + defer c.maxFindByIDsMu.Unlock() + + if c.maxFindByIDsLoaded { + return c.maxFindByIDs, nil + } + + response, err := c.ForgeClient.Version(ctx, &corev1.VersionRequest{DisplayConfig: true}) + if err != nil { + return 0, fmt.Errorf("get Core runtime config: %w", err) + } + c.maxFindByIDs = response.GetRuntimeConfig().GetMaxFindByIds() + c.maxFindByIDsLoaded = true + return c.maxFindByIDs, nil +} + +// findByIDBatches loads the shared server limit, fetches every chunk in order, +// and returns only a complete aggregate. The caller's context covers limit +// discovery and all batch RPCs; deadline exhaustion therefore fails the whole +// lookup without returning values from earlier batches. +func findByIDBatches[T any]( + ctx context.Context, + loadLimit func(context.Context) (uint32, error), + ids []string, + fetch func(context.Context, []string) ([]T, error), +) ([]T, error) { + if len(ids) == 0 { + return nil, nil + } + + limit, err := loadLimit(ctx) + if err != nil { + return nil, err + } + + batchSize := len(ids) + if limit > 0 && uint64(limit) < uint64(batchSize) { + batchSize = int(limit) + } + + result := make([]T, 0, len(ids)) + for batch := range slices.Chunk(ids, batchSize) { + values, err := fetch(ctx, batch) + if err != nil { + return nil, err + } + result = append(result, values...) + } + return result, nil +} + +// validateByIDsResponse rejects a batch response that omits any requested ID. +func validateByIDsResponse(requested, returned []string, rpcName string) error { + returnedSet := make(map[string]struct{}, len(returned)) + for _, id := range returned { + returnedSet[id] = struct{}{} + } + + missing := make([]string, 0) + for _, id := range requested { + if _, ok := returnedSet[id]; !ok { + missing = append(missing, id) + } + } + if len(missing) > 0 { + return fmt.Errorf("%s returned an incomplete response; missing IDs: %s", rpcName, strings.Join(missing, ", ")) + } + return nil +} + +// protoID is the common generated-protobuf ID contract used by Core resources. +type protoID interface { + GetId() string +} + +// protoIDsToStrings extracts ID values while preserving request order. +func protoIDsToStrings[T protoID](ids []T) []string { + result := make([]string, 0, len(ids)) + for _, id := range ids { + result = append(result, id.GetId()) + } + return result +} + +// fetchMachinesByIDs clones the caller's request for one raw Core batch and +// verifies that Core returned every requested machine. +func (c *batchingForgeClient) fetchMachinesByIDs( + ctx context.Context, + request *corev1.MachinesByIdsRequest, + batch []string, + options ...grpc.CallOption, +) ([]*corev1.Machine, error) { + batchRequest := proto.Clone(request).(*corev1.MachinesByIdsRequest) + batchRequest.MachineIds = stringsToMachineIds(batch) + response, err := c.ForgeClient.FindMachinesByIds(ctx, batchRequest, options...) + if err != nil { + return nil, fmt.Errorf("FindMachinesByIds: %w", err) + } + + machines := response.GetMachines() + returned := make([]string, 0, len(machines)) + for _, machine := range machines { + returned = append(returned, machine.GetId().GetId()) + } + if err := validateByIDsResponse(batch, returned, "FindMachinesByIds"); err != nil { + return nil, err + } + return machines, nil +} + // GetPowerStates returns the power states of the given machines (all machines if given an empty machineIds) func (c *grpcClient) GetPowerStates(ctx context.Context, machineIds []string) (ret []MachinePowerState, err error) { ctx, cancel := context.WithTimeout(ctx, c.grpcTimeout) @@ -332,6 +515,66 @@ func (c *grpcClient) FindHostMachineIdsByRack(ctx context.Context, rackID string return ids, nil } +// fetchSwitchesByIDs clones the caller's request for one raw Core batch and +// verifies that Core returned every requested switch. +func (c *batchingForgeClient) fetchSwitchesByIDs( + ctx context.Context, + request *corev1.SwitchesByIdsRequest, + batch []string, + options ...grpc.CallOption, +) ([]*corev1.Switch, error) { + batchRequest := proto.Clone(request).(*corev1.SwitchesByIdsRequest) + batchRequest.SwitchIds = make([]*corev1.SwitchId, 0, len(batch)) + for _, id := range batch { + batchRequest.SwitchIds = append(batchRequest.SwitchIds, &corev1.SwitchId{Id: id}) + } + + response, err := c.ForgeClient.FindSwitchesByIds(ctx, batchRequest, options...) + if err != nil { + return nil, fmt.Errorf("FindSwitchesByIds: %w", err) + } + + switches := response.GetSwitches() + returned := make([]string, 0, len(switches)) + for _, sw := range switches { + returned = append(returned, sw.GetId().GetId()) + } + if err := validateByIDsResponse(batch, returned, "FindSwitchesByIds"); err != nil { + return nil, err + } + return switches, nil +} + +// fetchPowerShelvesByIDs clones the caller's request for one raw Core batch and +// verifies that Core returned every requested power shelf. +func (c *batchingForgeClient) fetchPowerShelvesByIDs( + ctx context.Context, + request *corev1.PowerShelvesByIdsRequest, + batch []string, + options ...grpc.CallOption, +) ([]*corev1.PowerShelf, error) { + batchRequest := proto.Clone(request).(*corev1.PowerShelvesByIdsRequest) + batchRequest.PowerShelfIds = make([]*corev1.PowerShelfId, 0, len(batch)) + for _, id := range batch { + batchRequest.PowerShelfIds = append(batchRequest.PowerShelfIds, &corev1.PowerShelfId{Id: id}) + } + + response, err := c.ForgeClient.FindPowerShelvesByIds(ctx, batchRequest, options...) + if err != nil { + return nil, fmt.Errorf("FindPowerShelvesByIds: %w", err) + } + + shelves := response.GetPowerShelves() + returned := make([]string, 0, len(shelves)) + for _, shelf := range shelves { + returned = append(returned, shelf.GetId().GetId()) + } + if err := validateByIDsResponse(batch, returned, "FindPowerShelvesByIds"); err != nil { + return nil, err + } + return shelves, nil +} + // FindSwitchRackIDs returns the rack assignment of each given switch. func (c *grpcClient) FindSwitchRackIDs(ctx context.Context, switchIds []string) (map[string]string, error) { if len(switchIds) == 0 { diff --git a/rest-api/flow/internal/nicoapi/grpc_batch_test.go b/rest-api/flow/internal/nicoapi/grpc_batch_test.go new file mode 100644 index 0000000000..5f1efef645 --- /dev/null +++ b/rest-api/flow/internal/nicoapi/grpc_batch_test.go @@ -0,0 +1,412 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package nicoapi + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + corev1 "github.com/NVIDIA/infra-controller/rest-api/proto/core/gen/v1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" +) + +type recordingForgeClient struct { + corev1.ForgeClient + + mu sync.Mutex + runtimeConfig *corev1.RuntimeConfig + versionErr error + versionErrors []error + versionDelay time.Duration + switchDelay time.Duration + versionRequests []*corev1.VersionRequest + machineIDs []string + machineBatches [][]string + switchBatches [][]string + shelfBatches [][]string + failCall int + omitID string +} + +func (c *recordingForgeClient) Version( + ctx context.Context, + request *corev1.VersionRequest, + _ ...grpc.CallOption, +) (*corev1.BuildInfo, error) { + c.mu.Lock() + c.versionRequests = append(c.versionRequests, request) + call := len(c.versionRequests) + err := c.versionErr + if call <= len(c.versionErrors) { + err = c.versionErrors[call-1] + } + delay := c.versionDelay + config := c.runtimeConfig + c.mu.Unlock() + + if delay > 0 { + select { + case <-time.After(delay): + case <-ctx.Done(): + return nil, ctx.Err() + } + } + if err != nil { + return nil, err + } + return &corev1.BuildInfo{BuildVersion: "test", RuntimeConfig: config}, nil +} + +func (c *recordingForgeClient) FindMachineIds( + _ context.Context, + _ *corev1.MachineSearchConfig, + _ ...grpc.CallOption, +) (*corev1.MachineIdList, error) { + return &corev1.MachineIdList{MachineIds: stringsToMachineIds(c.machineIDs)}, nil +} + +func (c *recordingForgeClient) FindMachinesByIds( + _ context.Context, + request *corev1.MachinesByIdsRequest, + _ ...grpc.CallOption, +) (*corev1.MachineList, error) { + batch := protoIDsToStrings(request.GetMachineIds()) + c.mu.Lock() + c.machineBatches = append(c.machineBatches, batch) + failed := c.failCall == len(c.machineBatches) + omitID := c.omitID + c.mu.Unlock() + if failed { + return nil, errors.New("injected machine lookup failure") + } + + machines := make([]*corev1.Machine, 0, len(batch)) + for _, id := range batch { + if id != omitID { + machines = append(machines, &corev1.Machine{Id: &corev1.MachineId{Id: id}}) + } + } + return &corev1.MachineList{Machines: machines}, nil +} + +func (c *recordingForgeClient) FindSwitchesByIds( + ctx context.Context, + request *corev1.SwitchesByIdsRequest, + _ ...grpc.CallOption, +) (*corev1.SwitchList, error) { + batch := protoIDsToStrings(request.GetSwitchIds()) + c.mu.Lock() + c.switchBatches = append(c.switchBatches, batch) + failed := c.failCall == len(c.switchBatches) + omitID := c.omitID + delay := c.switchDelay + c.mu.Unlock() + if delay > 0 { + select { + case <-time.After(delay): + case <-ctx.Done(): + return nil, ctx.Err() + } + } + if failed { + return nil, errors.New("injected switch lookup failure") + } + + switches := make([]*corev1.Switch, 0, len(batch)) + for _, id := range batch { + if id != omitID { + nvosIP := "ip-" + id + switches = append(switches, &corev1.Switch{ + Id: &corev1.SwitchId{Id: id}, + RackId: &corev1.RackId{Id: "rack-" + id}, + ControllerState: "state-" + id, + NvosInfo: &corev1.SwitchNvosInfo{Ip: &nvosIP}, + }) + } + } + return &corev1.SwitchList{Switches: switches}, nil +} + +func (c *recordingForgeClient) FindPowerShelvesByIds( + _ context.Context, + request *corev1.PowerShelvesByIdsRequest, + _ ...grpc.CallOption, +) (*corev1.PowerShelfList, error) { + batch := protoIDsToStrings(request.GetPowerShelfIds()) + c.mu.Lock() + c.shelfBatches = append(c.shelfBatches, batch) + failed := c.failCall == len(c.shelfBatches) + omitID := c.omitID + c.mu.Unlock() + if failed { + return nil, errors.New("injected power shelf lookup failure") + } + + shelves := make([]*corev1.PowerShelf, 0, len(batch)) + for _, id := range batch { + if id != omitID { + shelves = append(shelves, &corev1.PowerShelf{ + Id: &corev1.PowerShelfId{Id: id}, + RackId: &corev1.RackId{Id: "rack-" + id}, + ControllerState: "state-" + id, + }) + } + } + return &corev1.PowerShelfList{PowerShelves: shelves}, nil +} + +func newRecordingGRPCClient(fake *recordingForgeClient) *grpcClient { + return &grpcClient{gclient: newBatchingForgeClient(fake), grpcTimeout: time.Second} +} + +func TestGrpcClient_ByIDLookupsHonorCoreBatchLimit(t *testing.T) { + ids := []string{"a", "b", "c", "d", "e"} + expectedBatches := [][]string{{"a", "b"}, {"c", "d"}, {"e"}} + + tests := []struct { + name string + invoke func(context.Context, *grpcClient, []string) (int, error) + batchCalls func(*recordingForgeClient) [][]string + }{ + { + name: "listed machines", + invoke: func(ctx context.Context, client *grpcClient, _ []string) (int, error) { + machines, err := client.GetMachines(ctx) + return len(machines), err + }, + batchCalls: func(fake *recordingForgeClient) [][]string { return fake.machineBatches }, + }, + { + name: "machines by IDs", + invoke: func(ctx context.Context, client *grpcClient, ids []string) (int, error) { + machines, err := client.FindMachinesByIds(ctx, ids) + return len(machines), err + }, + batchCalls: func(fake *recordingForgeClient) [][]string { return fake.machineBatches }, + }, + { + name: "direct switch lookup", + invoke: func(ctx context.Context, client *grpcClient, ids []string) (int, error) { + request := &corev1.SwitchesByIdsRequest{ + SwitchIds: make([]*corev1.SwitchId, 0, len(ids)), + } + for _, id := range ids { + request.SwitchIds = append(request.SwitchIds, &corev1.SwitchId{Id: id}) + } + response, err := client.gclient.FindSwitchesByIds(ctx, request) + return len(response.GetSwitches()), err + }, + batchCalls: func(fake *recordingForgeClient) [][]string { return fake.switchBatches }, + }, + { + name: "switch rack IDs", + invoke: func(ctx context.Context, client *grpcClient, ids []string) (int, error) { + values, err := client.FindSwitchRackIDs(ctx, ids) + return len(values), err + }, + batchCalls: func(fake *recordingForgeClient) [][]string { return fake.switchBatches }, + }, + { + name: "switch controller states", + invoke: func(ctx context.Context, client *grpcClient, ids []string) (int, error) { + values, err := client.FindSwitchControllerStates(ctx, ids) + return len(values), err + }, + batchCalls: func(fake *recordingForgeClient) [][]string { return fake.switchBatches }, + }, + { + name: "switch NVOS IPs", + invoke: func(ctx context.Context, client *grpcClient, ids []string) (int, error) { + values, err := client.FindSwitchNvosIPs(ctx, ids) + return len(values), err + }, + batchCalls: func(fake *recordingForgeClient) [][]string { return fake.switchBatches }, + }, + { + name: "power shelf rack IDs", + invoke: func(ctx context.Context, client *grpcClient, ids []string) (int, error) { + values, err := client.FindPowerShelfRackIDs(ctx, ids) + return len(values), err + }, + batchCalls: func(fake *recordingForgeClient) [][]string { return fake.shelfBatches }, + }, + { + name: "power shelf controller states", + invoke: func(ctx context.Context, client *grpcClient, ids []string) (int, error) { + values, err := client.FindPowerShelfControllerStates(ctx, ids) + return len(values), err + }, + batchCalls: func(fake *recordingForgeClient) [][]string { return fake.shelfBatches }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + fake := &recordingForgeClient{ + runtimeConfig: &corev1.RuntimeConfig{MaxFindByIds: 2}, + machineIDs: ids, + } + count, err := test.invoke(context.Background(), newRecordingGRPCClient(fake), ids) + + require.NoError(t, err) + assert.Equal(t, len(ids), count) + assert.Equal(t, expectedBatches, test.batchCalls(fake)) + require.Len(t, fake.versionRequests, 1) + assert.True(t, fake.versionRequests[0].GetDisplayConfig()) + }) + } +} + +func TestGrpcClient_ByIDLookupsTreatZeroOrAbsentLimitAsUnlimited(t *testing.T) { + ids := []string{"a", "b", "c"} + tests := []struct { + name string + config *corev1.RuntimeConfig + }{ + {name: "zero", config: &corev1.RuntimeConfig{MaxFindByIds: 0}}, + {name: "absent", config: nil}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + fake := &recordingForgeClient{runtimeConfig: test.config} + result, err := newRecordingGRPCClient(fake).FindSwitchRackIDs(context.Background(), ids) + + require.NoError(t, err) + assert.Len(t, result, len(ids)) + assert.Equal(t, [][]string{ids}, fake.switchBatches) + }) + } +} + +func TestGrpcClient_ByIDLookupsRejectPartialResults(t *testing.T) { + ids := []string{"a", "b", "c", "d"} + tests := []struct { + name string + fake *recordingForgeClient + errorString string + }{ + { + name: "runtime config lookup fails", + fake: &recordingForgeClient{ + versionErr: errors.New("injected Version failure"), + }, + errorString: "injected Version failure", + }, + { + name: "batch RPC fails", + fake: &recordingForgeClient{ + runtimeConfig: &corev1.RuntimeConfig{MaxFindByIds: 2}, + failCall: 2, + }, + errorString: "injected switch lookup failure", + }, + { + name: "batch response is incomplete", + fake: &recordingForgeClient{ + runtimeConfig: &corev1.RuntimeConfig{MaxFindByIds: 2}, + omitID: "c", + }, + errorString: "incomplete response; missing IDs: c", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + result, err := newRecordingGRPCClient(test.fake).FindSwitchRackIDs(context.Background(), ids) + + require.Error(t, err) + assert.ErrorContains(t, err, test.errorString) + assert.Nil(t, result) + }) + } +} + +func TestGrpcClient_ByIDLookupsCacheSuccessfulCoreLimit(t *testing.T) { + fake := &recordingForgeClient{ + runtimeConfig: &corev1.RuntimeConfig{MaxFindByIds: 2}, + } + client := newRecordingGRPCClient(fake) + + _, err := client.FindSwitchRackIDs(context.Background(), []string{"s1", "s2", "s3"}) + require.NoError(t, err) + _, err = client.FindPowerShelfRackIDs(context.Background(), []string{"p1", "p2", "p3"}) + require.NoError(t, err) + + require.Len(t, fake.versionRequests, 1, "all resource lookups must share the cached Core limit") +} + +func TestGrpcClient_ByIDLookupsRetryFailedCoreLimitLoad(t *testing.T) { + fake := &recordingForgeClient{ + runtimeConfig: &corev1.RuntimeConfig{MaxFindByIds: 2}, + versionErrors: []error{ + errors.New("transient Version failure"), + nil, + }, + } + client := newRecordingGRPCClient(fake) + + result, err := client.FindSwitchRackIDs(context.Background(), []string{"a", "b", "c"}) + require.ErrorContains(t, err, "transient Version failure") + assert.Nil(t, result) + + result, err = client.FindSwitchRackIDs(context.Background(), []string{"a", "b", "c"}) + require.NoError(t, err) + assert.Len(t, result, 3) + require.Len(t, fake.versionRequests, 2, "a failed limit load must not be cached") +} + +func TestGrpcClient_ByIDLookupsCoalesceConcurrentCoreLimitLoads(t *testing.T) { + fake := &recordingForgeClient{ + runtimeConfig: &corev1.RuntimeConfig{MaxFindByIds: 2}, + versionDelay: 20 * time.Millisecond, + } + client := newRecordingGRPCClient(fake) + + const callers = 8 + start := make(chan struct{}) + errs := make(chan error, callers) + var wg sync.WaitGroup + for range callers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + _, err := client.FindSwitchRackIDs(context.Background(), []string{"a", "b", "c"}) + errs <- err + }() + } + close(start) + wg.Wait() + close(errs) + + for err := range errs { + require.NoError(t, err) + } + require.Len(t, fake.versionRequests, 1, "concurrent first lookups must share one Version RPC") +} + +// One public lookup deadline covers the limit discovery and every batch. If +// earlier RPCs consume that budget, a later batch fails the whole lookup and +// no partial result is returned. +func TestGrpcClient_ByIDLookupsFailWithoutPartialResultWhenDeadlineStarvesLaterBatch(t *testing.T) { + fake := &recordingForgeClient{ + runtimeConfig: &corev1.RuntimeConfig{MaxFindByIds: 1}, + versionDelay: 10 * time.Millisecond, + switchDelay: 15 * time.Millisecond, + } + client := newRecordingGRPCClient(fake) + client.grpcTimeout = 35 * time.Millisecond + + result, err := client.FindSwitchRackIDs(context.Background(), []string{"a", "b", "c"}) + + require.ErrorIs(t, err, context.DeadlineExceeded) + assert.Nil(t, result) + assert.Less(t, len(fake.switchBatches), 3, "the exhausted operation deadline must prevent all batches from completing") +} diff --git a/rest-api/flow/internal/scheduler/jobs/inventorysync/expected_mirror_component.go b/rest-api/flow/internal/scheduler/jobs/inventorysync/expected_mirror_component.go index 5079077c6a..5e365ed2e1 100644 --- a/rest-api/flow/internal/scheduler/jobs/inventorysync/expected_mirror_component.go +++ b/rest-api/flow/internal/scheduler/jobs/inventorysync/expected_mirror_component.go @@ -437,15 +437,12 @@ func mirrorExpectedComponents( macs := hostBMCMACs(c) key := naturalKeyOrEmpty(c.Manufacturer, c.SerialNumber) if len(macs) == 0 && key == "" { - // Nothing on this row can be compared with what Core reported, so - // "Core dropped it" is indistinguishable from "Flow cannot - // recognise it". Left in place, with a warn so the operator can - // repair the row. - result.legacyExempt++ - log.Warn(). - Str("type", componentType). - Str("component_id", c.ID.String()). - Msg("Expected-inventory mirror: Flow component has neither a host BMC nor a manufacturer/serial pair; it cannot be matched against Core and is left in place") + // With the expected mirror enabled, a successful Core response is the + // authoritative snapshot for this component type. This row has no host + // BMC or complete fallback identity with which to join that snapshot; + // keeping it live would leave stale inventory visible to list/count and + // actual-drift paths. Remove it with the other absent expected rows. + p.toDelete = append(p.toDelete, *c) continue } if stillReportedByCore(macs, key, seenMACs, seenNaturalKeys) { @@ -464,6 +461,7 @@ func mirrorExpectedComponents( } now := time.Now() + softDeleted := 0 if err := pool.RunInTx(ctx, func(ctx context.Context, tx bun.Tx) error { for i := range p.toInsert { if _, err := tx.NewInsert().Model(&p.toInsert[i]).Exec(ctx); err != nil { @@ -524,9 +522,18 @@ func mirrorExpectedComponents( } } for i := range p.toDelete { - if _, err := tx.NewDelete().Model(&p.toDelete[i]).Where("id = ?", p.toDelete[i].ID).Exec(ctx); err != nil { + deleteResult, err := tx.NewDelete().Model(&p.toDelete[i]).Where("id = ?", p.toDelete[i].ID).Exec(ctx) + if err != nil { return fmt.Errorf("soft-delete component %q: %w", p.toDelete[i].SerialNumber, err) } + rowsAffected, err := deleteResult.RowsAffected() + if err != nil { + return fmt.Errorf("count soft-deleted component %q: %w", p.toDelete[i].SerialNumber, err) + } + if rowsAffected != 1 { + return fmt.Errorf("soft-delete component %q affected %d rows, expected 1", p.toDelete[i].SerialNumber, rowsAffected) + } + softDeleted += int(rowsAffected) } return nil }); err != nil { @@ -543,7 +550,7 @@ func mirrorExpectedComponents( result.inserted = len(p.toInsert) result.updated = len(p.toUpdate) - result.softDeleted = len(p.toDelete) + result.softDeleted = softDeleted return result } diff --git a/rest-api/flow/internal/scheduler/jobs/inventorysync/expected_mirror_db_test.go b/rest-api/flow/internal/scheduler/jobs/inventorysync/expected_mirror_db_test.go index d351878176..e71c311b1c 100644 --- a/rest-api/flow/internal/scheduler/jobs/inventorysync/expected_mirror_db_test.go +++ b/rest-api/flow/internal/scheduler/jobs/inventorysync/expected_mirror_db_test.go @@ -84,25 +84,63 @@ func computeSpec(mfr, serial, mac string) expectedComponentSpec { // --- rack mirror ---------------------------------------------------------- -// #11: a successful but empty Core response soft-deletes mirror-adopted racks, -// while legacy NULL-external_id racks are exempted. -func TestMirrorRacks_EmptyCoreSoftDeletesAdoptedNotLegacy(t *testing.T) { +// A successful but empty Core response soft-deletes both mirror-adopted racks +// and identity-less legacy rows. An identifiable legacy row remains available +// for later natural-key adoption. +func TestMirrorRacks_EmptyCoreDeletesAbsentAndIdentitylessRows(t *testing.T) { ctx, pool := mirrorTestPool(t) adopted := model.Rack{Name: "adopted", Manufacturer: "Mfg", SerialNumber: "AD-1", ExternalID: strPtr("a12")} require.NoError(t, adopted.Create(ctx, pool.DB)) - legacy := model.Rack{Name: "legacy", Manufacturer: "Mfg", SerialNumber: "LG-1"} - require.NoError(t, legacy.Create(ctx, pool.DB)) + identifiableLegacy := model.Rack{Name: "identifiable-legacy", Manufacturer: "Mfg", SerialNumber: "LG-1"} + require.NoError(t, identifiableLegacy.Create(ctx, pool.DB)) + identitylessLegacy := []model.Rack{ + {Name: "identityless-legacy"}, + {Name: "manufacturer-only-legacy", Manufacturer: "Mfg"}, + {Name: "serial-only-legacy", SerialNumber: "LG-2"}, + } + for i := range identitylessLegacy { + require.NoError(t, identitylessLegacy[i].Create(ctx, pool.DB)) + } - mirrorExpectedRacks(ctx, pool, nil) + result := mirrorExpectedRacks(ctx, pool, nil) + assert.Equal(t, 4, result.softDeleted, "summary count must reflect the four rows actually soft-deleted") gotAdopted, err := (&model.Rack{ID: adopted.ID}).GetIncludingDeleted(ctx, pool.DB) require.NoError(t, err) assert.NotNil(t, gotAdopted.DeletedAt, "adopted rack absent from Core must be soft-deleted") - gotLegacy, err := (&model.Rack{ID: legacy.ID}).GetIncludingDeleted(ctx, pool.DB) + gotIdentifiableLegacy, err := (&model.Rack{ID: identifiableLegacy.ID}).GetIncludingDeleted(ctx, pool.DB) require.NoError(t, err) - assert.Nil(t, gotLegacy.DeletedAt, "legacy NULL-external_id rack must be exempt from mirror deletes") + assert.Nil(t, gotIdentifiableLegacy.DeletedAt, "legacy rack with a complete natural key must remain available for later adoption") + + for _, legacy := range identitylessLegacy { + gotIdentitylessLegacy, err := (&model.Rack{ID: legacy.ID}).GetIncludingDeleted(ctx, pool.DB) + require.NoError(t, err) + assert.NotNil(t, gotIdentitylessLegacy.DeletedAt, "legacy rack without a complete identity must be soft-deleted") + } +} + +// An identity-less legacy rack can reserve a globally unique name needed by a +// Core rack. The first authoritative pass removes the orphan; the next pass can +// collect that tombstone and mirror the Core rack under the released name. +func TestMirrorRacks_IdentitylessRowCleanupReleasesNameForCoreRack(t *testing.T) { + ctx, pool := mirrorTestPool(t) + + orphan := model.Rack{Name: "reserved-name"} + require.NoError(t, orphan.Create(ctx, pool.DB)) + core := coreRackNamed("a12", "reserved-name", "Mfg", "CORE-1") + + first := mirrorExpectedRacks(ctx, pool, []nicoapi.ExpectedRackDetail{core}) + assert.Equal(t, 1, first.skippedNameTaken) + assert.Equal(t, 1, first.softDeleted) + + second := mirrorExpectedRacks(ctx, pool, []nicoapi.ExpectedRackDetail{core}) + assert.Equal(t, 1, second.inserted) + + var mirrored model.Rack + require.NoError(t, pool.DB.NewSelect().Model(&mirrored).Where("external_id = ?", "a12").Scan(ctx)) + assert.Equal(t, "reserved-name", mirrored.Name) } // #1: a soft-deleted rack is resurrected (deleted_at cleared) when Core @@ -362,7 +400,8 @@ func TestMirrorComponents_EmptyCoreSoftDeletesAll(t *testing.T) { c := model.Component{Type: compType(), Manufacturer: "Mfg", SerialNumber: "C-DEL-1"} require.NoError(t, c.Create(ctx, pool.DB)) - mirrorExpectedComponents(ctx, pool, compType(), nil, map[string]uuid.UUID{}) + result := mirrorExpectedComponents(ctx, pool, compType(), nil, map[string]uuid.UUID{}) + assert.Equal(t, 1, result.softDeleted, "summary count must reflect the row actually soft-deleted") got, err := (&model.Component{ID: c.ID}).GetIncludingDeleted(ctx, pool.DB) require.NoError(t, err) @@ -606,17 +645,27 @@ func TestMirrorComponents_BMCBoardSwapAdoptsByNaturalKey(t *testing.T) { assert.Equal(t, "aa:bb:cc:dd:ee:62", got.BMCs[0].MacAddress, "the host BMC must be repointed to Core's new MAC") } -// A Flow component with neither a host BMC nor a complete chassis pair can't be -// compared with what Core reported, so the delete phase leaves it alone. -func TestMirrorComponents_UnmatchableRowExemptFromDelete(t *testing.T) { +// A Flow component with neither a host BMC nor a complete chassis pair cannot +// join the successful authoritative snapshot, so reconciliation removes it +// rather than retaining stale live inventory. +func TestMirrorComponents_UnmatchableRowSoftDeleted(t *testing.T) { ctx, pool := mirrorTestPool(t) - c := model.Component{Type: compType(), Name: "orphan"} - require.NoError(t, c.Create(ctx, pool.DB)) + components := []model.Component{ + {Type: compType(), Name: "orphan"}, + {Type: compType(), Name: "manufacturer-only-orphan", Manufacturer: "Mfg"}, + {Type: compType(), Name: "serial-only-orphan", SerialNumber: "C-ORPHAN"}, + } + for i := range components { + require.NoError(t, components[i].Create(ctx, pool.DB)) + } - mirrorExpectedComponents(ctx, pool, compType(), nil, map[string]uuid.UUID{}) + result := mirrorExpectedComponents(ctx, pool, compType(), nil, map[string]uuid.UUID{}) + assert.Equal(t, 3, result.softDeleted, "summary count must reflect the three rows actually soft-deleted") - got, err := (&model.Component{ID: c.ID}).GetIncludingDeleted(ctx, pool.DB) - require.NoError(t, err) - assert.Nil(t, got.DeletedAt, "a component the mirror cannot match must not be soft-deleted") + for _, component := range components { + got, err := (&model.Component{ID: component.ID}).GetIncludingDeleted(ctx, pool.DB) + require.NoError(t, err) + assert.NotNil(t, got.DeletedAt, "a component without a complete expected-inventory identity must be soft-deleted") + } } diff --git a/rest-api/flow/internal/scheduler/jobs/inventorysync/expected_mirror_rack.go b/rest-api/flow/internal/scheduler/jobs/inventorysync/expected_mirror_rack.go index 05fc082a4a..1011f4dd05 100644 --- a/rest-api/flow/internal/scheduler/jobs/inventorysync/expected_mirror_rack.go +++ b/rest-api/flow/internal/scheduler/jobs/inventorysync/expected_mirror_rack.go @@ -75,10 +75,11 @@ func pullExpectedRacks( // 3. Live Flow rows whose external_id is set but no longer appear in Core // are soft-deleted (including the case where Core returned zero racks — // the caller only invokes this after a successful RPC, so empty is -// authoritative). Soft-deleted rows Core doesn't report either are left -// alone (already gone). Rows with a NULL external_id (legacy -// ingestion-gRPC rows the mirror has never adopted) are exempted and -// warn-logged so the operator has a visible signal of pending cleanup. +// authoritative). A legacy row with neither external_id nor a complete +// (manufacturer, serial_number) identity is also soft-deleted because no +// future snapshot can correlate it. An unmatched legacy row with a +// complete natural key remains exempt and is warn-logged for migration. +// Soft-deleted rows Core doesn't report are left alone (already gone). // // All writes for one pass happen in a single transaction so partial failures // can't leave the table half-mirrored. @@ -289,8 +290,9 @@ func mirrorExpectedRacks( // Reconcile the delete side. Already soft-deleted rows are skipped: if // Core still lists them, the match path above resurrected them; if not, // they're correctly gone already. Live Flow rows whose external_id is set - // but absent from Core get soft-deleted; legacy (NULL external_id) rows - // are exempted with a warn so the operator notices. + // but absent from Core get soft-deleted. Legacy rows without any complete + // identity are also deleted; identifiable legacy rows remain eligible for + // later natural-key adoption. for i := range flowRacks { r := &flowRacks[i] if r.DeletedAt != nil { @@ -311,18 +313,24 @@ func mirrorExpectedRacks( p.toDelete = append(p.toDelete, *r) continue } - // External_id is NULL — never adopted. Only legacy-warn if the - // (manufacturer, serial) doesn't appear in Core's set either, - // otherwise it'll be picked up by the adoption path above and a - // "future GC" warn would be misleading. A rack with an incomplete pair - // keys to the empty string, which is never in the set. - if _, adoptable := coreNaturalKeys[naturalKeyOrEmpty(r.Manufacturer, r.SerialNumber)]; !adoptable { + // External_id is NULL — never adopted. Without a complete natural key, + // this row cannot join the successful authoritative Core snapshot. Keeping + // it live also reserves its globally unique rack name, which can prevent a + // real Core rack from being mirrored. A complete but currently unmatched + // natural key remains a migration-compatible legacy row and may still be + // adopted by a later snapshot. + key := naturalKeyOrEmpty(r.Manufacturer, r.SerialNumber) + if key == "" { + p.toDelete = append(p.toDelete, *r) + continue + } + if _, adoptable := coreNaturalKeys[key]; !adoptable { result.legacyExempt++ log.Warn(). Str("rack_name", r.Name). Str("rack_serial", r.SerialNumber). Str("rack_manufacturer", r.Manufacturer). - Msg("Expected-inventory mirror: legacy Flow rack not present in Core's expected inventory; left in place for now (a follow-up will GC these once all sites have migrated)") + Msg("Expected-inventory mirror: identifiable legacy Flow rack not present in Core's expected inventory; left in place for possible later adoption") } } @@ -331,6 +339,7 @@ func mirrorExpectedRacks( } now := time.Now() + softDeleted := 0 if err := pool.RunInTx(ctx, func(ctx context.Context, tx bun.Tx) error { for i := range p.toInsert { if err := gcTombstoneForNameReuse(ctx, tx, tombstonesByName, p.toInsert[i].Name, uuid.Nil); err != nil { @@ -361,9 +370,18 @@ func mirrorExpectedRacks( } } for i := range p.toDelete { - if _, err := tx.NewDelete().Model(&p.toDelete[i]).Where("id = ?", p.toDelete[i].ID).Exec(ctx); err != nil { + deleteResult, err := tx.NewDelete().Model(&p.toDelete[i]).Where("id = ?", p.toDelete[i].ID).Exec(ctx) + if err != nil { return fmt.Errorf("soft-delete rack %q: %w", p.toDelete[i].Name, err) } + rowsAffected, err := deleteResult.RowsAffected() + if err != nil { + return fmt.Errorf("count soft-deleted rack %q: %w", p.toDelete[i].Name, err) + } + if rowsAffected != 1 { + return fmt.Errorf("soft-delete rack %q affected %d rows, expected 1", p.toDelete[i].Name, rowsAffected) + } + softDeleted += int(rowsAffected) } return nil }); err != nil { @@ -380,7 +398,7 @@ func mirrorExpectedRacks( result.inserted = len(p.toInsert) result.updated = len(p.toUpdate) - result.softDeleted = len(p.toDelete) + result.softDeleted = softDeleted return result }