From e58072f1aa0baf0eccf6d289112fb747da0f8379 Mon Sep 17 00:00:00 2001 From: Kun Zhao Date: Sun, 16 Aug 2026 18:03:31 -0700 Subject: [PATCH 1/4] fix(flow): honor Core find-by-ID limit Signed-off-by: Kun Zhao --- rest-api/flow/internal/nicoapi/grpc.go | 225 +++++++++++++- .../flow/internal/nicoapi/grpc_batch_test.go | 287 ++++++++++++++++++ 2 files changed, 511 insertions(+), 1 deletion(-) create mode 100644 rest-api/flow/internal/nicoapi/grpc_batch_test.go diff --git a/rest-api/flow/internal/nicoapi/grpc.go b/rest-api/flow/internal/nicoapi/grpc.go index 160fdceebe..f588003ef0 100644 --- a/rest-api/flow/internal/nicoapi/grpc.go +++ b/rest-api/flow/internal/nicoapi/grpc.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "os" + "strings" "sync" "testing" "time" @@ -19,6 +20,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 +56,56 @@ 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 +} + +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) { + client := &grpcClient{gclient: c.ForgeClient} + machines, err := client.findMachinesByRequest(ctx, request, 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) { + client := &grpcClient{gclient: c.ForgeClient} + switches, err := client.findSwitchesByRequest(ctx, request, 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) { + client := &grpcClient{gclient: c.ForgeClient} + shelves, err := client.findPowerShelvesByRequest(ctx, request, 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 +140,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 +240,114 @@ func (c *grpcClient) Version(ctx context.Context) (string, error) { return res.GetBuildVersion(), nil } +func (c *grpcClient) loadMaxFindByIDs(ctx context.Context) (uint32, error) { + response, err := c.gclient.Version(ctx, &corev1.VersionRequest{DisplayConfig: true}) + if err != nil { + return 0, fmt.Errorf("get Core runtime config: %w", err) + } + return response.GetRuntimeConfig().GetMaxFindByIds(), nil +} + +func findByIDBatches[T any]( + ctx context.Context, + client *grpcClient, + ids []string, + fetch func(context.Context, []string) ([]T, error), +) ([]T, error) { + if len(ids) == 0 { + return nil, nil + } + + limit, err := client.loadMaxFindByIDs(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 start := 0; start < len(ids); start += batchSize { + end := min(start+batchSize, len(ids)) + batch, err := fetch(ctx, ids[start:end]) + if err != nil { + return nil, err + } + result = append(result, batch...) + } + return result, nil +} + +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 +} + +func machineIDsToStrings(ids []*corev1.MachineId) []string { + result := make([]string, 0, len(ids)) + for _, id := range ids { + result = append(result, id.GetId()) + } + return result +} + +func switchIDsToStrings(ids []*corev1.SwitchId) []string { + result := make([]string, 0, len(ids)) + for _, id := range ids { + result = append(result, id.GetId()) + } + return result +} + +func powerShelfIDsToStrings(ids []*corev1.PowerShelfId) []string { + result := make([]string, 0, len(ids)) + for _, id := range ids { + result = append(result, id.GetId()) + } + return result +} + +func (c *grpcClient) findMachinesByRequest( + ctx context.Context, + request *corev1.MachinesByIdsRequest, + options ...grpc.CallOption, +) ([]*corev1.Machine, error) { + machineIDs := machineIDsToStrings(request.GetMachineIds()) + return findByIDBatches(ctx, c, machineIDs, func(ctx context.Context, batch []string) ([]*corev1.Machine, error) { + batchRequest := proto.Clone(request).(*corev1.MachinesByIdsRequest) + batchRequest.MachineIds = stringsToMachineIds(batch) + response, err := c.gclient.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 +495,66 @@ func (c *grpcClient) FindHostMachineIdsByRack(ctx context.Context, rackID string return ids, nil } +func (c *grpcClient) findSwitchesByRequest( + ctx context.Context, + request *corev1.SwitchesByIdsRequest, + options ...grpc.CallOption, +) ([]*corev1.Switch, error) { + switchIDs := switchIDsToStrings(request.GetSwitchIds()) + return findByIDBatches(ctx, c, switchIDs, func(ctx context.Context, batch []string) ([]*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.gclient.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 + }) +} + +func (c *grpcClient) findPowerShelvesByRequest( + ctx context.Context, + request *corev1.PowerShelvesByIdsRequest, + options ...grpc.CallOption, +) ([]*corev1.PowerShelf, error) { + shelfIDs := powerShelfIDsToStrings(request.GetPowerShelfIds()) + return findByIDBatches(ctx, c, shelfIDs, func(ctx context.Context, batch []string) ([]*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.gclient.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..213aa56bc1 --- /dev/null +++ b/rest-api/flow/internal/nicoapi/grpc_batch_test.go @@ -0,0 +1,287 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +package nicoapi + +import ( + "context" + "errors" + "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 + + runtimeConfig *corev1.RuntimeConfig + versionErr error + versionRequests []*corev1.VersionRequest + machineIDs []string + machineBatches [][]string + switchBatches [][]string + shelfBatches [][]string + failCall int + omitID string +} + +func (c *recordingForgeClient) Version( + _ context.Context, + request *corev1.VersionRequest, + _ ...grpc.CallOption, +) (*corev1.BuildInfo, error) { + c.versionRequests = append(c.versionRequests, request) + if c.versionErr != nil { + return nil, c.versionErr + } + return &corev1.BuildInfo{BuildVersion: "test", RuntimeConfig: c.runtimeConfig}, 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 := machineIDsToStrings(request.GetMachineIds()) + c.machineBatches = append(c.machineBatches, batch) + if c.failCall == len(c.machineBatches) { + return nil, errors.New("injected machine lookup failure") + } + + machines := make([]*corev1.Machine, 0, len(batch)) + for _, id := range batch { + if id != c.omitID { + machines = append(machines, &corev1.Machine{Id: &corev1.MachineId{Id: id}}) + } + } + return &corev1.MachineList{Machines: machines}, nil +} + +func (c *recordingForgeClient) FindSwitchesByIds( + _ context.Context, + request *corev1.SwitchesByIdsRequest, + _ ...grpc.CallOption, +) (*corev1.SwitchList, error) { + batch := switchIDsToStrings(request.GetSwitchIds()) + c.switchBatches = append(c.switchBatches, batch) + if c.failCall == len(c.switchBatches) { + return nil, errors.New("injected switch lookup failure") + } + + switches := make([]*corev1.Switch, 0, len(batch)) + for _, id := range batch { + if id != c.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 := powerShelfIDsToStrings(request.GetPowerShelfIds()) + c.shelfBatches = append(c.shelfBatches, batch) + if c.failCall == len(c.shelfBatches) { + return nil, errors.New("injected power shelf lookup failure") + } + + shelves := make([]*corev1.PowerShelf, 0, len(batch)) + for _, id := range batch { + if id != c.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) + }) + } +} From 2bf646bc945fb9cdd0ba4872ef6e3e275241e072 Mon Sep 17 00:00:00 2001 From: Kun Zhao Date: Sun, 16 Aug 2026 18:32:14 -0700 Subject: [PATCH 2/4] fix(flow): remove identity-less inventory rows Signed-off-by: Kun Zhao --- .../expected_mirror_component.go | 14 ++--- .../inventorysync/expected_mirror_db_test.go | 54 +++++++++++++------ .../inventorysync/expected_mirror_rack.go | 33 +++++++----- 3 files changed, 64 insertions(+), 37 deletions(-) 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..9319e9ea14 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,11 @@ 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") + // Nothing in this row can identify it in any future expected + // snapshot. Once the current snapshot succeeds, retaining the row + // would leave a permanent orphan, so remove it with the rest of the + // absent expected inventory. + p.toDelete = append(p.toDelete, *c) continue } if stillReportedByCore(macs, key, seenMACs, seenNaturalKeys) { 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..3fe4ae3549 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,15 +84,24 @@ 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) @@ -100,9 +109,15 @@ func TestMirrorRacks_EmptyCoreSoftDeletesAdoptedNotLegacy(t *testing.T) { 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") + } } // #1: a soft-deleted rack is resurrected (deleted_at cleared) when Core @@ -606,17 +621,26 @@ 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 +// be identified by any future expected snapshot, so a successful reconciliation +// removes it rather than retaining a permanent orphan. +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{}) - 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..7e3a574430 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,23 @@ 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, + // no future Core snapshot can identify this row, so the successful + // authoritative snapshot makes it safe to remove. 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") } } From 0f1760a7e07f77243f18fdf926aebcf4e7f6e98f Mon Sep 17 00:00:00 2001 From: Kun Zhao Date: Tue, 18 Aug 2026 10:51:30 -0700 Subject: [PATCH 3/4] refactor(flow): cache Core inventory batch limit Signed-off-by: Kun Zhao --- rest-api/flow/internal/nicoapi/grpc.go | 206 ++++++++++-------- .../flow/internal/nicoapi/grpc_batch_test.go | 153 +++++++++++-- 2 files changed, 252 insertions(+), 107 deletions(-) diff --git a/rest-api/flow/internal/nicoapi/grpc.go b/rest-api/flow/internal/nicoapi/grpc.go index f588003ef0..0cbac07ca3 100644 --- a/rest-api/flow/internal/nicoapi/grpc.go +++ b/rest-api/flow/internal/nicoapi/grpc.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "os" + "slices" "strings" "sync" "testing" @@ -61,6 +62,10 @@ type grpcClient struct { // methods in this package. type batchingForgeClient struct { corev1.ForgeClient + + maxFindByIDsMu sync.Mutex + maxFindByIDs uint32 + maxFindByIDsLoaded bool } func newBatchingForgeClient(client corev1.ForgeClient) corev1.ForgeClient { @@ -72,8 +77,11 @@ func (c *batchingForgeClient) FindMachinesByIds( request *corev1.MachinesByIdsRequest, options ...grpc.CallOption, ) (*corev1.MachineList, error) { - client := &grpcClient{gclient: c.ForgeClient} - machines, err := client.findMachinesByRequest(ctx, request, options...) + 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 } @@ -85,8 +93,11 @@ func (c *batchingForgeClient) FindSwitchesByIds( request *corev1.SwitchesByIdsRequest, options ...grpc.CallOption, ) (*corev1.SwitchList, error) { - client := &grpcClient{gclient: c.ForgeClient} - switches, err := client.findSwitchesByRequest(ctx, request, options...) + 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 } @@ -98,8 +109,11 @@ func (c *batchingForgeClient) FindPowerShelvesByIds( request *corev1.PowerShelvesByIdsRequest, options ...grpc.CallOption, ) (*corev1.PowerShelfList, error) { - client := &grpcClient{gclient: c.ForgeClient} - shelves, err := client.findPowerShelvesByRequest(ctx, request, options...) + 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 } @@ -240,17 +254,33 @@ func (c *grpcClient) Version(ctx context.Context) (string, error) { return res.GetBuildVersion(), nil } -func (c *grpcClient) loadMaxFindByIDs(ctx context.Context) (uint32, error) { - response, err := c.gclient.Version(ctx, &corev1.VersionRequest{DisplayConfig: true}) +// 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) } - return response.GetRuntimeConfig().GetMaxFindByIds(), nil + 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, - client *grpcClient, + loadLimit func(context.Context) (uint32, error), ids []string, fetch func(context.Context, []string) ([]T, error), ) ([]T, error) { @@ -258,7 +288,7 @@ func findByIDBatches[T any]( return nil, nil } - limit, err := client.loadMaxFindByIDs(ctx) + limit, err := loadLimit(ctx) if err != nil { return nil, err } @@ -269,17 +299,17 @@ func findByIDBatches[T any]( } result := make([]T, 0, len(ids)) - for start := 0; start < len(ids); start += batchSize { - end := min(start+batchSize, len(ids)) - batch, err := fetch(ctx, ids[start:end]) + for batch := range slices.Chunk(ids, batchSize) { + values, err := fetch(ctx, batch) if err != nil { return nil, err } - result = append(result, batch...) + 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 { @@ -298,23 +328,13 @@ func validateByIDsResponse(requested, returned []string, rpcName string) error { return nil } -func machineIDsToStrings(ids []*corev1.MachineId) []string { - result := make([]string, 0, len(ids)) - for _, id := range ids { - result = append(result, id.GetId()) - } - return result -} - -func switchIDsToStrings(ids []*corev1.SwitchId) []string { - result := make([]string, 0, len(ids)) - for _, id := range ids { - result = append(result, id.GetId()) - } - return result +// protoID is the common generated-protobuf ID contract used by Core resources. +type protoID interface { + GetId() string } -func powerShelfIDsToStrings(ids []*corev1.PowerShelfId) []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()) @@ -322,30 +342,30 @@ func powerShelfIDsToStrings(ids []*corev1.PowerShelfId) []string { return result } -func (c *grpcClient) findMachinesByRequest( +// 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) { - machineIDs := machineIDsToStrings(request.GetMachineIds()) - return findByIDBatches(ctx, c, machineIDs, func(ctx context.Context, batch []string) ([]*corev1.Machine, error) { - batchRequest := proto.Clone(request).(*corev1.MachinesByIdsRequest) - batchRequest.MachineIds = stringsToMachineIds(batch) - response, err := c.gclient.FindMachinesByIds(ctx, batchRequest, options...) - if err != nil { - return nil, fmt.Errorf("FindMachinesByIds: %w", err) - } + 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 - }) + 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) @@ -495,64 +515,64 @@ func (c *grpcClient) FindHostMachineIdsByRack(ctx context.Context, rackID string return ids, nil } -func (c *grpcClient) findSwitchesByRequest( +// 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) { - switchIDs := switchIDsToStrings(request.GetSwitchIds()) - return findByIDBatches(ctx, c, switchIDs, func(ctx context.Context, batch []string) ([]*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}) - } + 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.gclient.FindSwitchesByIds(ctx, batchRequest, options...) - if err != nil { - return nil, fmt.Errorf("FindSwitchesByIds: %w", err) - } + 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 - }) + 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 } -func (c *grpcClient) findPowerShelvesByRequest( +// 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) { - shelfIDs := powerShelfIDsToStrings(request.GetPowerShelfIds()) - return findByIDBatches(ctx, c, shelfIDs, func(ctx context.Context, batch []string) ([]*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}) - } + 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.gclient.FindPowerShelvesByIds(ctx, batchRequest, options...) - if err != nil { - return nil, fmt.Errorf("FindPowerShelvesByIds: %w", err) - } + 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 - }) + 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. diff --git a/rest-api/flow/internal/nicoapi/grpc_batch_test.go b/rest-api/flow/internal/nicoapi/grpc_batch_test.go index 213aa56bc1..5f1efef645 100644 --- a/rest-api/flow/internal/nicoapi/grpc_batch_test.go +++ b/rest-api/flow/internal/nicoapi/grpc_batch_test.go @@ -6,6 +6,7 @@ package nicoapi import ( "context" "errors" + "sync" "testing" "time" @@ -18,8 +19,12 @@ import ( 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 @@ -30,15 +35,32 @@ type recordingForgeClient struct { } func (c *recordingForgeClient) Version( - _ context.Context, + ctx context.Context, request *corev1.VersionRequest, _ ...grpc.CallOption, ) (*corev1.BuildInfo, error) { + c.mu.Lock() c.versionRequests = append(c.versionRequests, request) - if c.versionErr != nil { - return nil, c.versionErr + 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() + } } - return &corev1.BuildInfo{BuildVersion: "test", RuntimeConfig: c.runtimeConfig}, nil + if err != nil { + return nil, err + } + return &corev1.BuildInfo{BuildVersion: "test", RuntimeConfig: config}, nil } func (c *recordingForgeClient) FindMachineIds( @@ -54,15 +76,19 @@ func (c *recordingForgeClient) FindMachinesByIds( request *corev1.MachinesByIdsRequest, _ ...grpc.CallOption, ) (*corev1.MachineList, error) { - batch := machineIDsToStrings(request.GetMachineIds()) + batch := protoIDsToStrings(request.GetMachineIds()) + c.mu.Lock() c.machineBatches = append(c.machineBatches, batch) - if c.failCall == len(c.machineBatches) { + 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 != c.omitID { + if id != omitID { machines = append(machines, &corev1.Machine{Id: &corev1.MachineId{Id: id}}) } } @@ -70,19 +96,31 @@ func (c *recordingForgeClient) FindMachinesByIds( } func (c *recordingForgeClient) FindSwitchesByIds( - _ context.Context, + ctx context.Context, request *corev1.SwitchesByIdsRequest, _ ...grpc.CallOption, ) (*corev1.SwitchList, error) { - batch := switchIDsToStrings(request.GetSwitchIds()) + batch := protoIDsToStrings(request.GetSwitchIds()) + c.mu.Lock() c.switchBatches = append(c.switchBatches, batch) - if c.failCall == len(c.switchBatches) { + 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 != c.omitID { + if id != omitID { nvosIP := "ip-" + id switches = append(switches, &corev1.Switch{ Id: &corev1.SwitchId{Id: id}, @@ -100,15 +138,19 @@ func (c *recordingForgeClient) FindPowerShelvesByIds( request *corev1.PowerShelvesByIdsRequest, _ ...grpc.CallOption, ) (*corev1.PowerShelfList, error) { - batch := powerShelfIDsToStrings(request.GetPowerShelfIds()) + batch := protoIDsToStrings(request.GetPowerShelfIds()) + c.mu.Lock() c.shelfBatches = append(c.shelfBatches, batch) - if c.failCall == len(c.shelfBatches) { + 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 != c.omitID { + if id != omitID { shelves = append(shelves, &corev1.PowerShelf{ Id: &corev1.PowerShelfId{Id: id}, RackId: &corev1.RackId{Id: "rack-" + id}, @@ -285,3 +327,86 @@ func TestGrpcClient_ByIDLookupsRejectPartialResults(t *testing.T) { }) } } + +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") +} From be18d3a7e3db09430408d1c1b1688dad027a902b Mon Sep 17 00:00:00 2001 From: Kun Zhao Date: Tue, 18 Aug 2026 10:51:38 -0700 Subject: [PATCH 4/4] fix(flow): verify orphan inventory cleanup Signed-off-by: Kun Zhao --- .../expected_mirror_component.go | 23 ++++++++---- .../inventorysync/expected_mirror_db_test.go | 35 ++++++++++++++++--- .../inventorysync/expected_mirror_rack.go | 23 ++++++++---- 3 files changed, 64 insertions(+), 17 deletions(-) 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 9319e9ea14..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,10 +437,11 @@ func mirrorExpectedComponents( macs := hostBMCMACs(c) key := naturalKeyOrEmpty(c.Manufacturer, c.SerialNumber) if len(macs) == 0 && key == "" { - // Nothing in this row can identify it in any future expected - // snapshot. Once the current snapshot succeeds, retaining the row - // would leave a permanent orphan, so remove it with the rest of the - // absent expected inventory. + // 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 } @@ -460,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 { @@ -520,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 { @@ -539,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 3fe4ae3549..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 @@ -103,7 +103,8 @@ func TestMirrorRacks_EmptyCoreDeletesAbsentAndIdentitylessRows(t *testing.T) { 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) @@ -120,6 +121,28 @@ func TestMirrorRacks_EmptyCoreDeletesAbsentAndIdentitylessRows(t *testing.T) { } } +// 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 // re-reports it, keeping the UUID stable. func TestMirrorRacks_ResurrectOnReReport(t *testing.T) { @@ -377,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) @@ -622,8 +646,8 @@ func TestMirrorComponents_BMCBoardSwapAdoptsByNaturalKey(t *testing.T) { } // A Flow component with neither a host BMC nor a complete chassis pair cannot -// be identified by any future expected snapshot, so a successful reconciliation -// removes it rather than retaining a permanent orphan. +// 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) @@ -636,7 +660,8 @@ func TestMirrorComponents_UnmatchableRowSoftDeleted(t *testing.T) { 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") for _, component := range components { got, err := (&model.Component{ID: component.ID}).GetIncludingDeleted(ctx, pool.DB) 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 7e3a574430..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 @@ -314,10 +314,11 @@ func mirrorExpectedRacks( continue } // External_id is NULL — never adopted. Without a complete natural key, - // no future Core snapshot can identify this row, so the successful - // authoritative snapshot makes it safe to remove. A complete but - // currently unmatched natural key remains a migration-compatible legacy - // row and may still be adopted by a later snapshot. + // 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) @@ -338,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 { @@ -368,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 { @@ -387,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 }