Remove outer transaction and use controller

Using the registration methods directly caused issues with only partly
removing various objects from UFS, notably leaving orphaned machines
when removing asset entries.
Remove the outer transaction from the sync and re-add the controller
methods that account for removing all objects as needed from UFS.

TEST=make test
Bug: b:475260738
Change-Id: Ib2ae8a11e4fadd6d4ca63f1367c9b7224b09085f

Change-Id: Ia5f40ac8fb56c76f105fcc0d8afd8d3aeadb7ecb
Reviewed-on: https://chromium-review.googlesource.com/c/infra/infra/+/7477344
Auto-Submit: Warren Whitman <warrenpw@google.com>
Commit-Queue: Gowri Denduluri <gowriden@google.com>
Reviewed-by: Gowri Denduluri <gowriden@google.com>
Cr-Commit-Position: refs/heads/main@{#78416}

go/src/infra/unifiedfleet/app/dumper/nlyte_os_bq_sync.go

diff --git a/go/src/infra/unifiedfleet/app/dumper/nlyte_os_bq_sync.go b/go/src/infra/unifiedfleet/app/dumper/nlyte_os_bq_sync.go
index 892d65a..f117b6c 100644
--- a/go/src/infra/unifiedfleet/app/dumper/nlyte_os_bq_sync.go
+++ b/go/src/infra/unifiedfleet/app/dumper/nlyte_os_bq_sync.go
@@ -18,7 +18,6 @@
 	"go.chromium.org/luci/common/errors"
 	"go.chromium.org/luci/common/logging"
 	"go.chromium.org/luci/common/sync/parallel"
-	"go.chromium.org/luci/gae/service/datastore"
 
 	ufspb "go.chromium.org/infra/unifiedfleet/api/v1/models"
 	"go.chromium.org/infra/unifiedfleet/app/config"
@@ -181,6 +180,7 @@
 			"zone": {zoneCfg.GetZone()},
 		}
 		var getAssetsFn func(context.Context, int32, string, map[string][]any, bool) ([]*ufspb.Asset, string, error)
+		// When listing assets we can use registration as we aren't trying to modify any entries in Datastore.
 		switch zoneCfg.GetDefaultSyncAssetType() {
 		case config.NlyteSyncConfig_ZoneConfig_NLYTE_ASSET:
 			getAssetsFn = registration.ListNlyteAssets
@@ -203,7 +203,8 @@
 				merr = append(merr, errors.Fmt("listing zone assets for removal: %w", err))
 				break
 			}
-			err = datastore.RunInTransaction(ctx, genRemoveStaleZoneAssetsFunc(nlyteAssetMap, assets, zoneCfg), nil)
+			// Don't use a transaction here as it is already implemented per action in controller
+			err := genRemoveStaleZoneAssetsFunc(nlyteAssetMap, assets, zoneCfg)(ctx)
 			if err != nil {
 				logging.Errorf(ctx, "Failure removing stale zone assets: %v", err)
 				merr = append(merr, errors.Fmt("removing stale zone assets: %w", err))
@@ -299,10 +300,10 @@
 		switch zoneCfg.GetDefaultSyncAssetType() {
 		case config.NlyteSyncConfig_ZoneConfig_NLYTE_ASSET:
 			logging.Debugf(ctx, "Zone asset sync type is NLYTE_ASSET; using registration.DeleteNlyteAsset")
-			zoneRmFunc = registration.DeleteNlyteAsset
+			zoneRmFunc = controller.DeleteNlyteAsset
 		case config.NlyteSyncConfig_ZoneConfig_ASSET:
 			logging.Debugf(ctx, "Zone asset sync type is ASSET; using registration.DeleteAsset")
-			zoneRmFunc = registration.DeleteAsset
+			zoneRmFunc = controller.DeleteAsset
 		default:
 			return errors.New("unknown asset sync type")
 		}
@@ -322,10 +323,10 @@
 			switch rowConfig.GetRowSyncAssetType() {
 			case config.NlyteSyncConfig_ZoneConfig_NLYTE_ASSET:
 				logging.Debugf(ctx, "Row asset sync type is NLYTE_ASSET; using registration.DeleteNlyteAsset")
-				rmFunc = registration.DeleteNlyteAsset
+				rmFunc = controller.DeleteNlyteAsset
 			case config.NlyteSyncConfig_ZoneConfig_ASSET:
 				logging.Debugf(ctx, "Row asset sync type is ASSET; using registration.DeleteAsset")
-				rmFunc = registration.DeleteAsset
+				rmFunc = controller.DeleteAsset
 			case config.NlyteSyncConfig_ZoneConfig_WRITE_ASSET_AS_UNSPECIFIED:
 				logging.Debugf(ctx, "Using zone default asset sync type for asset %q of row %q", asset.GetName(), asset.GetLocation().GetRow())
 				rmFunc = zoneRmFunc