PublicStore支持直传

client/internal/cmdline/test.go

@@ -1,293 +1,187 @@
 package cmdline
 
-/*
 import (
 	"context"
 	"fmt"
-	"io"
+	"os"
+	"time"
 
 	"github.com/spf13/cobra"
-	"gitlink.org.cn/cloudream/common/pkgs/future"
 	"gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec"
-	cdssdk "gitlink.org.cn/cloudream/common/sdks/storage"
-	"gitlink.org.cn/cloudream/common/utils/math2"
+	"gitlink.org.cn/cloudream/common/pkgs/logger"
+	"gitlink.org.cn/cloudream/jcs-pub/client/internal/accessstat"
+	"gitlink.org.cn/cloudream/jcs-pub/client/internal/config"
+	"gitlink.org.cn/cloudream/jcs-pub/client/internal/db"
+	"gitlink.org.cn/cloudream/jcs-pub/client/internal/downloader"
+	"gitlink.org.cn/cloudream/jcs-pub/client/internal/downloader/strategy"
+	"gitlink.org.cn/cloudream/jcs-pub/client/internal/metacache"
+	"gitlink.org.cn/cloudream/jcs-pub/client/internal/services"
+	"gitlink.org.cn/cloudream/jcs-pub/client/internal/uploader"
 	stgglb "gitlink.org.cn/cloudream/jcs-pub/common/globals"
+	"gitlink.org.cn/cloudream/jcs-pub/common/models/datamap"
+	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/connectivity"
+	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/distlock"
 	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2"
 	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2/parser"
-	coormq "gitlink.org.cn/cloudream/jcs-pub/common/pkgs/mq/coordinator"
+	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/pool"
+	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/sysevent"
 )
 
 func init() {
-	RootCmd.AddCommand(&cobra.Command{
+	var configPath string
+	cmd := cobra.Command{
 		Use:   "test",
 		Short: "test",
 		Run: func(cmd *cobra.Command, args []string) {
-			coorCli, err := stgglb.CoordinatorMQPool.Acquire()
-			if err != nil {
-				panic(err)
-			}
-			defer stgglb.CoordinatorMQPool.Release(coorCli)
-
-			stgs, err := coorCli.GetStorageDetails(coormq.ReqGetStorageDetails([]cdssdk.StorageID{1, 2, 3, 4, 5}))
-			if err != nil {
-				panic(err)
-			}
-
-			ft := ioswitch2.NewFromTo()
-			ft.ECParam = cdssdk.NewECRedundancy(3, 6, 1024*1024*5)
-			// ft.SegmentParam = cdssdk.NewSegmentRedundancy(1024*100*3, 3)
-			ft.AddFrom(ioswitch2.NewFromShardstore("FullC036CBB7553A909F8B8877D4461924307F27ECB66CFF928EEEAFD569C3887E29", *stgs.Storages[3].MasterHub, *stgs.Storages[3], ioswitch2.ECStream(2)))
-			ft.AddFrom(ioswitch2.NewFromShardstore("Full543F38D9F524238AC0239263AA0DD4B4328763818EA98A7A5F72E59748FDA27A", *stgs.Storages[3].MasterHub, *stgs.Storages[3], ioswitch2.ECStream(3)))
-			ft.AddFrom(ioswitch2.NewFromShardstore("Full50B464DB2FDDC29D0380D9FFAB6D944FAF5C7624955D757939280590F01F3ECD", *stgs.Storages[3].MasterHub, *stgs.Storages[3], ioswitch2.ECStream(4)))
-			// ft.AddFrom(ioswitch2.NewFromShardstore("Full4D142C458F2399175232D5636235B09A84664D60869E925EB20FFBE931045BDD", *stgs.Storages[1].MasterHub, *stgs.Storages[1], ioswitch2.SegmentStream(2)))
-			// ft.AddFrom(ioswitch2.NewFromShardstore("Full03B5CF4B57251D7BB4308FE5C81AF5A21E2B28994CC7CB1FB37698DAE271DC22", *stgs.Storages[2].MasterHub, *stgs.Storages[2], ioswitch2.RawStream()))
-			// ft.AddTo(ioswitch2.NewToShardStore(*stgs.Storages[3].MasterHub, *stgs.Storages[3], ioswitch2.RawStream(), "0"))
-			// ft.AddTo(ioswitch2.NewToShardStore(*stgs.Storages[1].MasterHub, stgs.Storages[1].Storage, ioswitch2.SegmentStream(0), "0"))
-			// ft.AddTo(ioswitch2.NewToShardStoreWithRange(*stgs.Storages[1].MasterHub, *stgs.Storages[1], ioswitch2.SegmentStream(1), "1", math2.Range{Offset: 1}))
-			ft.AddTo(ioswitch2.NewToShardStore(*stgs.Storages[4].MasterHub, *stgs.Storages[4], ioswitch2.ECStream(0), "0"))
-			ft.AddTo(ioswitch2.NewToShardStore(*stgs.Storages[4].MasterHub, *stgs.Storages[4], ioswitch2.ECStream(1), "1"))
-			ft.AddTo(ioswitch2.NewToShardStore(*stgs.Storages[4].MasterHub, *stgs.Storages[4], ioswitch2.ECStream(5), "2"))
-
-			err = parser.Parse(ft, plans)
-			if err != nil {
-				panic(err)
-			}
-
-			fmt.Printf("plans: %v\n", plans)
-
-			exec := plans.Execute(exec.NewExecContext())
-
-			fut := future.NewSetVoid()
-			go func() {
-				mp, err := exec.Wait(context.Background())
-				if err != nil {
-					panic(err)
-				}
-
-				fmt.Printf("0: %v, 1: %v, 2: %v\n", mp["0"], mp["1"], mp["2"])
-				fut.SetVoid()
-			}()
-
-			fut.Wait(context.TODO())
+			test(configPath)
 		},
-	})
-
-	RootCmd.AddCommand(&cobra.Command{
-		Use:   "test32",
-		Short: "test32",
-		Run: func(cmd *cobra.Command, args []string) {
-			coorCli, err := stgglb.CoordinatorMQPool.Acquire()
-			if err != nil {
-				panic(err)
-			}
-			defer stgglb.CoordinatorMQPool.Release(coorCli)
-
-			stgs, err := coorCli.GetStorageDetails(coormq.ReqGetStorageDetails([]cdssdk.StorageID{1, 2}))
-			if err != nil {
-				panic(err)
-			}
-
-			ft := ioswitch2.NewFromTo()
-			ft.SegmentParam = cdssdk.NewSegmentRedundancy(1293, 3)
-			ft.AddFrom(ioswitch2.NewFromShardstore("4E69A8B8CD9F42EDE371DA94458BADFB2308AFCA736AA393784A3D81F4746377", *stgs.Storages[0].MasterHub, *stgs.Storages[0], ioswitch2.RawStream()))
-			// ft.AddTo(ioswitch2.NewToShardStore(*stgs.Storages[1].MasterHub, stgs.Storages[1].Storage, ioswitch2.SegmentStream(0), "0"))
-			ft.AddTo(ioswitch2.NewToShardStoreWithRange(*stgs.Storages[1].MasterHub, *stgs.Storages[1], ioswitch2.SegmentStream(1), "1", math2.Range{Offset: 1}))
-			ft.AddTo(ioswitch2.NewToShardStore(*stgs.Storages[1].MasterHub, *stgs.Storages[1], ioswitch2.SegmentStream(2), "2"))
-
-			plans := exec.NewPlanBuilder()
-			err = parser.Parse(ft, plans)
-			if err != nil {
-				panic(err)
-			}
-
-			fmt.Printf("plans: %v\n", plans)
-
-			exec := plans.Execute(exec.NewExecContext())
-
-			fut := future.NewSetVoid()
-			go func() {
-				mp, err := exec.Wait(context.Background())
-				if err != nil {
-					panic(err)
-				}
-
-				fmt.Printf("0: %v, 1: %v, 2: %v\n", mp["0"], mp["1"], mp["2"])
-				fut.SetVoid()
-			}()
-
-			fut.Wait(context.TODO())
-		},
-	})
-	RootCmd.AddCommand(&cobra.Command{
-		Use:   "test1",
-		Short: "test1",
-		Run: func(cmd *cobra.Command, args []string) {
-			coorCli, err := stgglb.CoordinatorMQPool.Acquire()
-			if err != nil {
-				panic(err)
-			}
-			defer stgglb.CoordinatorMQPool.Release(coorCli)
-
-			stgs, err := coorCli.GetStorageDetails(coormq.ReqGetStorageDetails([]cdssdk.StorageID{1, 2}))
-			if err != nil {
-				panic(err)
-			}
-
-			ft := ioswitch2.NewFromTo()
-			ft.SegmentParam = cdssdk.NewSegmentRedundancy(1293, 3)
-			ft.ECParam = &cdssdk.DefaultECRedundancy
-			ft.AddFrom(ioswitch2.NewFromShardstore("22CC59CE3297F78F2D20DC1E33181B77F21E6782097C94E1664F99F129834069", *stgs.Storages[1].MasterHub, *stgs.Storages[1], ioswitch2.SegmentStream(0)))
-			ft.AddFrom(ioswitch2.NewFromShardstore("5EAC20EB3EBC7B5FA176C5BD1C01041FB2A6D14C35D6A232CA83D7F1E4B01ADE", *stgs.Storages[1].MasterHub, *stgs.Storages[1], ioswitch2.SegmentStream(1)))
-			ft.AddFrom(ioswitch2.NewFromShardstore("A9BC1802F37100C80C72A1D6E8F53C0E0B73F85F99153D8C78FB01CEC9D8D903", *stgs.Storages[1].MasterHub, *stgs.Storages[1], ioswitch2.SegmentStream(2)))
-
-			toDrv, drvStr := ioswitch2.NewToDriverWithRange(ioswitch2.RawStream(), math2.NewRange(0, 1293))
-			ft.AddTo(toDrv)
-			ft.AddTo(ioswitch2.NewToShardStore(*stgs.Storages[0].MasterHub, *stgs.Storages[0], ioswitch2.ECStream(0), "EC0"))
-			ft.AddTo(ioswitch2.NewToShardStore(*stgs.Storages[0].MasterHub, *stgs.Storages[0], ioswitch2.ECStream(1), "EC1"))
-			ft.AddTo(ioswitch2.NewToShardStore(*stgs.Storages[0].MasterHub, *stgs.Storages[0], ioswitch2.ECStream(2), "EC2"))
-
-			plans := exec.NewPlanBuilder()
-			err = parser.Parse(ft, plans)
-			if err != nil {
-				panic(err)
-			}
+	}
+	cmd.Flags().StringVarP(&configPath, "config", "c", "", "config file path")
+	RootCmd.AddCommand(&cmd)
+}
 
-			fmt.Printf("plans: %v\n", plans)
+func doTest(svc *services.Service) {
+	ft := ioswitch2.NewFromTo()
 
-			exec := plans.Execute(exec.NewExecContext())
+	space1 := svc.UserSpaceMeta.Get(3)
+	space2 := svc.UserSpaceMeta.Get(4)
 
-			fut := future.NewSetVoid()
-			go func() {
-				mp, err := exec.Wait(context.Background())
-				if err != nil {
-					panic(err)
-				}
+	// ft.AddFrom(ioswitch2.NewFromPublicStore(*space1.MasterHub, *space1, "space3/blocks/1A/Full1AE5436AF72D8EF93923486E0E167315CEF0C91898064DADFAC22216FFBC5E3D"))
+	// ft.AddTo(ioswitch2.NewToPublicStore(*space2.MasterHub, *space2, "block"))
+	// plans := exec.NewPlanBuilder()
+	// parser.Parse(ft, plans)
+	// fmt.Println(plans)
 
-				for k, v := range mp {
-					fmt.Printf("%s: %v\n", k, v)
-				}
+	// _, err := plans.Execute(exec.NewExecContext()).Wait(context.Background())
+	// fmt.Println(err)
 
-				fut.SetVoid()
-			}()
-
-			go func() {
-				str, err := exec.BeginRead(drvStr)
-				if err != nil {
-					panic(err)
-				}
+	ft = ioswitch2.NewFromTo()
+	ft.AddFrom(ioswitch2.NewFromShardstore("Full1AE5436AF72D8EF93923486E0E167315CEF0C91898064DADFAC22216FFBC5E3D", *space1.MasterHub, *space1, ioswitch2.RawStream()))
+	ft.AddTo(ioswitch2.NewToPublicStore(*space2.MasterHub, *space2, "test3.txt"))
+	plans := exec.NewPlanBuilder()
+	parser.Parse(ft, plans)
+	fmt.Println(plans)
+	_, err := plans.Execute(exec.NewExecContext()).Wait(context.Background())
+	fmt.Println(err)
 
-				data, err := io.ReadAll(str)
-				if err != nil {
-					panic(err)
-				}
-
-				fmt.Printf("read(%v): %s\n", len(data), string(data))
-			}()
+}
 
-			fut.Wait(context.TODO())
-		},
+func test(configPath string) {
+	err := config.Init(configPath)
+	if err != nil {
+		fmt.Printf("init config failed, err: %s", err.Error())
+		os.Exit(1)
+	}
+
+	err = logger.Init(&config.Cfg().Logger)
+	if err != nil {
+		fmt.Printf("init logger failed, err: %s", err.Error())
+		os.Exit(1)
+	}
+
+	stgglb.InitLocal(config.Cfg().Local)
+	stgglb.InitMQPool(config.Cfg().RabbitMQ)
+	stgglb.InitHubRPCPool(&config.Cfg().HubGRPC)
+
+	// 数据库
+	db, err := db.NewDB(&config.Cfg().DB)
+	if err != nil {
+		logger.Fatalf("new db failed, err: %s", err.Error())
+	}
+
+	// 初始化系统事件发布器
+	evtPub, err := sysevent.NewPublisher(sysevent.ConfigFromMQConfig(config.Cfg().RabbitMQ), &datamap.SourceClient{
+		UserID: config.Cfg().Local.UserID,
 	})
-
-	RootCmd.AddCommand(&cobra.Command{
-		Use:   "test4",
-		Short: "test4",
-		Run: func(cmd *cobra.Command, args []string) {
-			coorCli, err := stgglb.CoordinatorMQPool.Acquire()
-			if err != nil {
-				panic(err)
+	if err != nil {
+		logger.Errorf("new sysevent publisher: %v", err)
+		os.Exit(1)
+	}
+	evtPubChan := evtPub.Start()
+	defer evtPub.Stop()
+
+	// 连接性信息收集
+	conCol := connectivity.NewCollector(&config.Cfg().Connectivity, nil)
+	conCol.CollecNow()
+
+	// 元数据缓存
+	metaCacheHost := metacache.NewHost(db)
+	go metaCacheHost.Serve()
+	stgMeta := metaCacheHost.AddStorageMeta()
+	hubMeta := metaCacheHost.AddHubMeta()
+	conMeta := metaCacheHost.AddConnectivity()
+
+	// 公共锁
+	publock := distlock.NewService()
+
+	// 访问统计
+	acStat := accessstat.NewAccessStat(accessstat.Config{
+		// TODO 考虑放到配置里
+		ReportInterval: time.Second * 10,
+	}, db)
+	acStatChan := acStat.Start()
+	defer acStat.Stop()
+
+	// 存储管理器
+	stgPool := pool.NewPool()
+
+	// 下载策略
+	strgSel := strategy.NewSelector(config.Cfg().DownloadStrategy, stgMeta, hubMeta, conMeta)
+
+	// 下载器
+	dlder := downloader.NewDownloader(config.Cfg().Downloader, &conCol, stgPool, strgSel, db)
+
+	// 上传器
+	uploader := uploader.NewUploader(publock, &conCol, stgPool, stgMeta, db)
+
+	svc := services.NewService(publock, dlder, acStat, uploader, strgSel, stgMeta, db, evtPub, nil)
+
+	go func() {
+		doTest(svc)
+		os.Exit(0)
+	}()
+	/// 开始监听各个模块的事件
+
+	evtPubEvt := evtPubChan.Receive()
+	acStatEvt := acStatChan.Receive()
+
+loop:
+	for {
+		select {
+		case e := <-evtPubEvt.Chan():
+			if e.Err != nil {
+				logger.Errorf("receive publisher event: %v", err)
+				break loop
 			}
-			defer stgglb.CoordinatorMQPool.Release(coorCli)
-
-			stgs, err := coorCli.GetStorageDetails(coormq.ReqGetStorageDetails([]cdssdk.StorageID{1, 2}))
-			if err != nil {
-				panic(err)
-			}
-
-			ft := ioswitch2.NewFromTo()
-			ft.ECParam = &cdssdk.DefaultECRedundancy
-			ft.AddFrom(ioswitch2.NewFromShardstore("4E69A8B8CD9F42EDE371DA94458BADFB2308AFCA736AA393784A3D81F4746377", *stgs.Storages[1].MasterHub, *stgs.Storages[1], ioswitch2.RawStream()))
-			ft.AddTo(ioswitch2.NewToShardStore(*stgs.Storages[0].MasterHub, *stgs.Storages[0], ioswitch2.ECStream(0), "EC0"))
-			ft.AddTo(ioswitch2.NewToShardStore(*stgs.Storages[0].MasterHub, *stgs.Storages[0], ioswitch2.ECStream(1), "EC1"))
-			ft.AddTo(ioswitch2.NewToShardStore(*stgs.Storages[0].MasterHub, *stgs.Storages[0], ioswitch2.ECStream(2), "EC2"))
-
-			plans := exec.NewPlanBuilder()
-			err = parser.Parse(ft, plans)
-			if err != nil {
-				panic(err)
-			}
-
-			fmt.Printf("plans: %v\n", plans)
-
-			exec := plans.Execute(exec.NewExecContext())
 
-			fut := future.NewSetVoid()
-			go func() {
-				mp, err := exec.Wait(context.Background())
-				if err != nil {
-					panic(err)
-				}
+			switch val := e.Value.(type) {
+			case sysevent.PublishError:
+				logger.Errorf("publishing event: %v", val)
 
-				for k, v := range mp {
-					fmt.Printf("%s: %v\n", k, v)
+			case sysevent.PublisherExited:
+				if val.Err != nil {
+					logger.Errorf("publisher exited with error: %v", val.Err)
+				} else {
+					logger.Info("publisher exited")
 				}
+				break loop
 
-				fut.SetVoid()
-			}()
-
-			fut.Wait(context.TODO())
-		},
-	})
-
-	RootCmd.AddCommand(&cobra.Command{
-	RootCmd.AddCommand(&cobra.Command{
-		Use:   "test11",
-		Short: "test11",
-		Run: func(cmd *cobra.Command, args []string) {
-			coorCli, err := stgglb.CoordinatorMQPool.Acquire()
-			if err != nil {
-				panic(err)
+			case sysevent.OtherError:
+				logger.Errorf("sysevent: %v", val)
 			}
-			defer stgglb.CoordinatorMQPool.Release(coorCli)
+			evtPubEvt = evtPubChan.Receive()
 
-			stgs, err := coorCli.GetStorageDetails(coormq.ReqGetStorageDetails([]cdssdk.StorageID{1, 2}))
-			if err != nil {
-				panic(err)
+		case e := <-acStatEvt.Chan():
+			if e.Err != nil {
+				logger.Errorf("receive access stat event: %v", err)
+				break loop
 			}
-
-			ft := ioswitch2.NewFromTo()
-			ft.SegmentParam = cdssdk.NewSegmentRedundancy(1293, 3)
-			ft.ECParam = &cdssdk.DefaultECRedundancy
-			ft.AddFrom(ioswitch2.NewFromShardstore("22CC59CE3297F78F2D20DC1E33181B77F21E6782097C94E1664F99F129834069", *stgs.Storages[1].MasterHub, *stgs.Storages[1], ioswitch2.SegmentStream(0)))
-			ft.AddFrom(ioswitch2.NewFromShardstore("5EAC20EB3EBC7B5FA176C5BD1C01041FB2A6D14C35D6A232CA83D7F1E4B01ADE", *stgs.Storages[1].MasterHub, *stgs.Storages[1], ioswitch2.SegmentStream(1)))
-			ft.AddFrom(ioswitch2.NewFromShardstore("A9BC1802F37100C80C72A1D6E8F53C0E0B73F85F99153D8C78FB01CEC9D8D903", *stgs.Storages[1].MasterHub, *stgs.Storages[1], ioswitch2.SegmentStream(2)))
-			ft.AddTo(ioswitch2.NewToShardStoreWithRange(*stgs.Storages[0].MasterHub, *stgs.Storages[0], ioswitch2.RawStream(), "raw", math2.NewRange(10, 645)))
-
-			plans := exec.NewPlanBuilder()
-			err = parser.Parse(ft, plans)
-			if err != nil {
-				panic(err)
+			switch e := e.Value.(type) {
+			case accessstat.ExitEvent:
+				logger.Infof("access stat exited, err: %v", e.Err)
+				break loop
 			}
-
-			fmt.Printf("plans: %v\n", plans)
-
-			exec := plans.Execute(exec.NewExecContext())
-
-			fut := future.NewSetVoid()
-			go func() {
-				mp, err := exec.Wait(context.Background())
-				if err != nil {
-					panic(err)
-				}
-
-				for k, v := range mp {
-					fmt.Printf("%s: %v\n", k, v)
-				}
-
-				fut.SetVoid()
-			}()
-
-			fut.Wait(context.TODO())
-		},
-	})
+			acStatEvt = acStatChan.Receive()
+		}
+	}
 }
-*/

common/pkgs/ioswitch2/fromto.go

@@ -112,7 +112,7 @@ func (f *FromDriver) GetStreamIndex() StreamIndex {
 type FromShardstore struct {
 	FileHash    clitypes.FileHash
 	Hub         cortypes.Hub
-	Space       clitypes.UserSpaceDetail
+	UserSpace   clitypes.UserSpaceDetail
 	StreamIndex StreamIndex
 }
 
@@ -120,7 +120,7 @@ func NewFromShardstore(fileHash clitypes.FileHash, hub cortypes.Hub, space clity
 	return &FromShardstore{
 		FileHash:    fileHash,
 		Hub:         hub,
-		Space:       space,
+		UserSpace:   space,
 		StreamIndex: strIdx,
 	}
 }
@@ -130,16 +130,16 @@ func (f *FromShardstore) GetStreamIndex() StreamIndex {
 }
 
 type FromPublicStore struct {
-	Hub   cortypes.Hub
-	Space clitypes.UserSpaceDetail
-	Path  string
+	Hub       cortypes.Hub
+	UserSpace clitypes.UserSpaceDetail
+	Path      string
 }
 
 func NewFromPublicStore(hub cortypes.Hub, space clitypes.UserSpaceDetail, path string) *FromPublicStore {
 	return &FromPublicStore{
-		Hub:   hub,
-		Space: space,
-		Path:  path,
+		Hub:       hub,
+		UserSpace: space,
+		Path:      path,
 	}
 }
 
@@ -215,26 +215,26 @@ func (t *ToShardStore) GetRange() math2.Range {
 	return t.Range
 }
 
-type LoadToPublic struct {
+type ToPublicStore struct {
 	Hub        cortypes.Hub
 	Space      clitypes.UserSpaceDetail
 	ObjectPath string
 }
 
-func NewToPublicStore(hub cortypes.Hub, space clitypes.UserSpaceDetail, objectPath string) *LoadToPublic {
-	return &LoadToPublic{
+func NewToPublicStore(hub cortypes.Hub, space clitypes.UserSpaceDetail, objectPath string) *ToPublicStore {
+	return &ToPublicStore{
 		Hub:        hub,
 		Space:      space,
 		ObjectPath: objectPath,
 	}
 }
 
-func (t *LoadToPublic) GetStreamIndex() StreamIndex {
+func (t *ToPublicStore) GetStreamIndex() StreamIndex {
 	return StreamIndex{
 		Type: StreamIndexRaw,
 	}
 }
 
-func (t *LoadToPublic) GetRange() math2.Range {
+func (t *ToPublicStore) GetRange() math2.Range {
 	return math2.Range{}
 }

common/pkgs/ioswitch2/ops2/bypass.go

@@ -12,23 +12,26 @@ import (
 
 func init() {
 	exec.UseOp[*BypassToShardStore]()
-	exec.UseVarValue[*BypassUploadedFileValue]()
+	exec.UseOp[*BypassToPublicStore]()
+
+	exec.UseVarValue[*BypassedFileInfoValue]()
 	exec.UseVarValue[*BypassHandleResultValue]()
 
 	exec.UseOp[*BypassFromShardStore]()
+	exec.UseOp[*BypassFromPublicStore]()
 	exec.UseVarValue[*BypassFilePathValue]()
 
 	exec.UseOp[*BypassFromShardStoreHTTP]()
 	exec.UseVarValue[*HTTPRequestValue]()
 }
 
-type BypassUploadedFileValue struct {
-	types.BypassUploadedFile
+type BypassedFileInfoValue struct {
+	types.BypassedFileInfo
 }
 
-func (v *BypassUploadedFileValue) Clone() exec.VarValue {
-	return &BypassUploadedFileValue{
-		BypassUploadedFile: v.BypassUploadedFile,
+func (v *BypassedFileInfoValue) Clone() exec.VarValue {
+	return &BypassedFileInfoValue{
+		BypassedFileInfo: v.BypassedFileInfo,
 	}
 }
 
@@ -46,7 +49,7 @@ type BypassToShardStore struct {
 	UserSpace      clitypes.UserSpaceDetail
 	BypassFileInfo exec.VarID
 	BypassCallback exec.VarID
-	FileHash       exec.VarID
+	FileInfo       exec.VarID
 }
 
 func (o *BypassToShardStore) Execute(ctx *exec.ExecContext, e *exec.Executor) error {
@@ -60,23 +63,23 @@ func (o *BypassToShardStore) Execute(ctx *exec.ExecContext, e *exec.Executor) er
 		return err
 	}
 
-	br, ok := shardStore.(types.BypassWrite)
+	br, ok := shardStore.(types.BypassShardWrite)
 	if !ok {
 		return fmt.Errorf("shard store %v not support bypass write", o.UserSpace)
 	}
 
-	fileInfo, err := exec.BindVar[*BypassUploadedFileValue](e, ctx.Context, o.BypassFileInfo)
+	fileInfo, err := exec.BindVar[*BypassedFileInfoValue](e, ctx.Context, o.BypassFileInfo)
 	if err != nil {
 		return err
 	}
 
-	err = br.BypassUploaded(fileInfo.BypassUploadedFile)
+	err = br.BypassedShard(fileInfo.BypassedFileInfo)
 	if err != nil {
 		return err
 	}
 
 	e.PutVar(o.BypassCallback, &BypassHandleResultValue{Commited: true})
-	e.PutVar(o.FileHash, &ShardInfoValue{Hash: fileInfo.Hash, Size: fileInfo.Size})
+	e.PutVar(o.FileInfo, &ShardInfoValue{Hash: fileInfo.Hash, Size: fileInfo.Size})
 	return nil
 }
 
@@ -84,6 +87,47 @@ func (o *BypassToShardStore) String() string {
 	return fmt.Sprintf("BypassToShardStore[UserSpace:%v] Info: %v, Callback: %v", o.UserSpace, o.BypassFileInfo, o.BypassCallback)
 }
 
+type BypassToPublicStore struct {
+	UserSpace      clitypes.UserSpaceDetail
+	BypassFileInfo exec.VarID
+	BypassCallback exec.VarID
+	DestPath       string
+}
+
+func (o *BypassToPublicStore) Execute(ctx *exec.ExecContext, e *exec.Executor) error {
+	stgPool, err := exec.GetValueByType[*pool.Pool](ctx)
+	if err != nil {
+		return err
+	}
+
+	store, err := stgPool.GetPublicStore(&o.UserSpace)
+	if err != nil {
+		return err
+	}
+
+	br, ok := store.(types.BypassPublicWrite)
+	if !ok {
+		return fmt.Errorf("public store %v not support bypass write", o.UserSpace)
+	}
+
+	fileInfo, err := exec.BindVar[*BypassedFileInfoValue](e, ctx.Context, o.BypassFileInfo)
+	if err != nil {
+		return err
+	}
+
+	err = br.BypassedPublic(fileInfo.BypassedFileInfo, o.DestPath)
+	if err != nil {
+		return err
+	}
+
+	e.PutVar(o.BypassCallback, &BypassHandleResultValue{Commited: true})
+	return nil
+}
+
+func (o *BypassToPublicStore) String() string {
+	return fmt.Sprintf("BypassToPublicStore[UserSpace:%v] Info: %v, Callback: %v", o.UserSpace, o.BypassFileInfo, o.BypassCallback)
+}
+
 type BypassFilePathValue struct {
 	types.BypassFilePath
 }
@@ -111,12 +155,12 @@ func (o *BypassFromShardStore) Execute(ctx *exec.ExecContext, e *exec.Executor)
 		return err
 	}
 
-	br, ok := shardStore.(types.BypassRead)
+	br, ok := shardStore.(types.BypassShardRead)
 	if !ok {
 		return fmt.Errorf("shard store %v not support bypass read", o.UserSpace)
 	}
 
-	path, err := br.BypassRead(o.FileHash)
+	path, err := br.BypassShardRead(o.FileHash)
 	if err != nil {
 		return err
 	}
@@ -129,6 +173,41 @@ func (o *BypassFromShardStore) String() string {
 	return fmt.Sprintf("BypassFromShardStore[UserSpace:%v] FileHash: %v, Output: %v", o.UserSpace, o.FileHash, o.Output)
 }
 
+type BypassFromPublicStore struct {
+	UserSpace clitypes.UserSpaceDetail
+	Path      string
+	Output    exec.VarID
+}
+
+func (o *BypassFromPublicStore) Execute(ctx *exec.ExecContext, e *exec.Executor) error {
+	stgPool, err := exec.GetValueByType[*pool.Pool](ctx)
+	if err != nil {
+		return err
+	}
+
+	store, err := stgPool.GetPublicStore(&o.UserSpace)
+	if err != nil {
+		return err
+	}
+
+	br, ok := store.(types.BypassPublicRead)
+	if !ok {
+		return fmt.Errorf("public store %v not support bypass read", o.UserSpace)
+	}
+
+	path, err := br.BypassPublicRead(o.Path)
+	if err != nil {
+		return err
+	}
+
+	e.PutVar(o.Output, &BypassFilePathValue{BypassFilePath: path})
+	return nil
+}
+
+func (o *BypassFromPublicStore) String() string {
+	return fmt.Sprintf("BypassFromPublicStore[UserSpace:%v] Path: %v, Output: %v", o.UserSpace, o.Path, o.Output)
+}
+
 // 旁路Http读取
 type BypassFromShardStoreHTTP struct {
 	UserSpace clitypes.UserSpaceDetail
@@ -157,7 +236,7 @@ func (o *BypassFromShardStoreHTTP) Execute(ctx *exec.ExecContext, e *exec.Execut
 		return err
 	}
 
-	br, ok := shardStore.(types.HTTPBypassRead)
+	br, ok := shardStore.(types.HTTPBypassShardRead)
 	if !ok {
 		return fmt.Errorf("shard store %v not support bypass read", o.UserSpace)
 	}
@@ -220,7 +299,48 @@ func (t *BypassToShardStoreNode) GenerateOp() (exec.Op, error) {
 		UserSpace:      t.UserSpace,
 		BypassFileInfo: t.BypassFileInfoSlot().Var().VarID,
 		BypassCallback: t.BypassCallbackVar().Var().VarID,
-		FileHash:       t.FileHashVar().Var().VarID,
+		FileInfo:       t.FileHashVar().Var().VarID,
+	}, nil
+}
+
+type BypassToPublicStoreNode struct {
+	dag.NodeBase
+	UserSpace clitypes.UserSpaceDetail
+	DestPath  string
+}
+
+func (b *GraphNodeBuilder) NewBypassToPublicStore(userSpace clitypes.UserSpaceDetail, dstPath string) *BypassToPublicStoreNode {
+	node := &BypassToPublicStoreNode{
+		UserSpace: userSpace,
+		DestPath:  dstPath,
+	}
+	b.AddNode(node)
+
+	node.InputValues().Init(1)
+	node.OutputValues().Init(node, 1)
+	return node
+}
+
+func (n *BypassToPublicStoreNode) BypassFileInfoSlot() dag.ValueInputSlot {
+	return dag.ValueInputSlot{
+		Node:  n,
+		Index: 0,
+	}
+}
+
+func (n *BypassToPublicStoreNode) BypassCallbackVar() dag.ValueOutputSlot {
+	return dag.ValueOutputSlot{
+		Node:  n,
+		Index: 0,
+	}
+}
+
+func (t *BypassToPublicStoreNode) GenerateOp() (exec.Op, error) {
+	return &BypassToPublicStore{
+		UserSpace:      t.UserSpace,
+		BypassFileInfo: t.BypassFileInfoSlot().Var().VarID,
+		BypassCallback: t.BypassCallbackVar().Var().VarID,
+		DestPath:       t.DestPath,
 	}, nil
 }
 
@@ -257,6 +377,38 @@ func (n *BypassFromShardStoreNode) GenerateOp() (exec.Op, error) {
 	}, nil
 }
 
+type BypassFromPublicStoreNode struct {
+	dag.NodeBase
+	UserSpace clitypes.UserSpaceDetail
+	Path      string
+}
+
+func (b *GraphNodeBuilder) NewBypassFromPublicStore(userSpace clitypes.UserSpaceDetail, path string) *BypassFromPublicStoreNode {
+	node := &BypassFromPublicStoreNode{
+		UserSpace: userSpace,
+		Path:      path,
+	}
+	b.AddNode(node)
+
+	node.OutputValues().Init(node, 1)
+	return node
+}
+
+func (n *BypassFromPublicStoreNode) FilePathVar() dag.ValueOutputSlot {
+	return dag.ValueOutputSlot{
+		Node:  n,
+		Index: 0,
+	}
+}
+
+func (n *BypassFromPublicStoreNode) GenerateOp() (exec.Op, error) {
+	return &BypassFromPublicStore{
+		UserSpace: n.UserSpace,
+		Path:      n.Path,
+		Output:    n.FilePathVar().Var().VarID,
+	}, nil
+}
+
 // 旁路Http读取
 type BypassFromShardStoreHTTPNode struct {
 	dag.NodeBase

common/pkgs/ioswitch2/ops2/ec.go

@@ -193,10 +193,10 @@ func (o *CallECMultiplier) Execute(ctx *exec.ExecContext, e *exec.Executor) erro
 	}
 	defer ecMul.Abort()
 
-	outputVals := make([]*BypassUploadedFileValue, 0, len(outputs))
+	outputVals := make([]*BypassedFileInfoValue, 0, len(outputs))
 	for _, output := range outputs {
-		outputVals = append(outputVals, &BypassUploadedFileValue{
-			BypassUploadedFile: output,
+		outputVals = append(outputVals, &BypassedFileInfoValue{
+			BypassedFileInfo: output,
 		})
 	}
 	exec.PutArray(e, o.Outputs, outputVals)

common/pkgs/ioswitch2/ops2/multipart.go

@@ -88,8 +88,8 @@ func (o *MultipartInitiator) Execute(ctx *exec.ExecContext, e *exec.Executor) er
 	}
 
 	// 告知后续Op临时文件的路径
-	e.PutVar(o.BypassFileOutput, &BypassUploadedFileValue{
-		BypassUploadedFile: fileInfo,
+	e.PutVar(o.BypassFileOutput, &BypassedFileInfoValue{
+		BypassedFileInfo: fileInfo,
 	})
 
 	// 等待后续Op处理临时文件

common/pkgs/ioswitch2/ops2/s2s.go

@@ -20,6 +20,7 @@ type S2STransfer struct {
 	Dst            clitypes.UserSpaceDetail
 	Output         exec.VarID
 	BypassCallback exec.VarID
+	S2SOption      types.S2SOption
 }
 
 func (o *S2STransfer) Execute(ctx *exec.ExecContext, e *exec.Executor) error {
@@ -39,14 +40,14 @@ func (o *S2STransfer) Execute(ctx *exec.ExecContext, e *exec.Executor) error {
 	}
 
 	// 传输文件
-	dstPath, err := s2s.Transfer(ctx.Context, &o.Src, srcPath.Path)
+	dstPath, err := s2s.Transfer(ctx.Context, &o.Src, srcPath.Path, o.S2SOption)
 	if err != nil {
 		return err
 	}
 	defer s2s.Abort()
 
 	// 告知后续Op处理临时文件
-	e.PutVar(o.Output, &BypassUploadedFileValue{BypassUploadedFile: types.BypassUploadedFile{
+	e.PutVar(o.Output, &BypassedFileInfoValue{BypassedFileInfo: types.BypassedFileInfo{
 		Path: dstPath,
 		Hash: srcPath.Info.Hash,
 		Size: srcPath.Info.Size,
@@ -66,19 +67,21 @@ func (o *S2STransfer) Execute(ctx *exec.ExecContext, e *exec.Executor) error {
 }
 
 func (o *S2STransfer) String() string {
-	return fmt.Sprintf("S2STransfer %v:%v -> %v:%v", o.Src.Storage.String(), o.SrcPath, o.Dst.Storage.String(), o.Output)
+	return fmt.Sprintf("S2STransfer %v:%v -> %v:%v, Callback: %v", o.Src.Storage.String(), o.SrcPath, o.Dst.Storage.String(), o.Output, o.BypassCallback)
 }
 
 type S2STransferNode struct {
 	dag.NodeBase
 	Src clitypes.UserSpaceDetail
 	Dst clitypes.UserSpaceDetail
+	Opt types.S2SOption
 }
 
-func (b *GraphNodeBuilder) NewS2STransfer(src, dst clitypes.UserSpaceDetail) *S2STransferNode {
+func (b *GraphNodeBuilder) NewS2STransfer(src, dst clitypes.UserSpaceDetail, opt types.S2SOption) *S2STransferNode {
 	n := &S2STransferNode{
 		Src: src,
 		Dst: dst,
+		Opt: opt,
 	}
 	b.AddNode(n)
 
@@ -116,5 +119,6 @@ func (n *S2STransferNode) GenerateOp() (exec.Op, error) {
 		Dst:            n.Dst,
 		Output:         n.BypassFileInfoVar().Var().VarID,
 		BypassCallback: n.BypassCallbackSlot().Var().VarID,
+		S2SOption:      n.Opt,
 	}, nil
 }

common/pkgs/ioswitch2/parser/gen/generator.go

@@ -259,7 +259,7 @@ func buildFromNode(ctx *state.GenerateState, f ioswitch2.From) (ops2.FromNode, e
 
 	switch f := f.(type) {
 	case *ioswitch2.FromShardstore:
-		t := ctx.DAG.NewShardRead(f, f.Space, types.NewOpen(f.FileHash))
+		t := ctx.DAG.NewShardRead(f, f.UserSpace, types.NewOpen(f.FileHash))
 
 		if f.StreamIndex.IsRaw() {
 			t.Open.WithNullableLength(repRange.Offset, repRange.Length)
@@ -338,7 +338,7 @@ func buildFromNode(ctx *state.GenerateState, f ioswitch2.From) (ops2.FromNode, e
 
 	case *ioswitch2.FromPublicStore:
 		// TODO 可以考虑支持设置读取范围
-		n := ctx.DAG.NewPublicRead(f, f.Space, f.Path)
+		n := ctx.DAG.NewPublicRead(f, f.UserSpace, f.Path)
 		switch addr := f.Hub.Address.(type) {
 		case *cortypes.HttpAddressInfo:
 			n.Env().ToEnvWorker(&ioswitch2.HttpHubWorker{Hub: f.Hub})
@@ -379,7 +379,7 @@ func buildToNode(ctx *state.GenerateState, t ioswitch2.To) (ops2.ToNode, error)
 
 		return n, nil
 
-	case *ioswitch2.LoadToPublic:
+	case *ioswitch2.ToPublicStore:
 		n := ctx.DAG.NewPublicWrite(t, t.Space, t.ObjectPath)
 
 		if err := setEnvByAddress(n, t.Hub, t.Hub.Address); err != nil {

common/pkgs/ioswitch2/parser/opt/ec.go

@@ -88,7 +88,7 @@ func UseECMultiplier(ctx *state.GenerateState) {
 				return true
 			}
 
-			if !factory.GetBuilder(&srNode.From.Space).FeatureDesc().HasBypassHTTPRead {
+			if !factory.GetBuilder(&srNode.From.UserSpace).FeatureDesc().HasBypassHTTPRead {
 				return true
 			}
 

common/pkgs/ioswitch2/parser/opt/s2s.go

@@ -5,6 +5,7 @@ import (
 	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2/ops2"
 	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/ioswitch2/parser/state"
 	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/factory"
+	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/types"
 )
 
 // 将直接从一个存储服务传到另一个存储服务的过程换成S2S传输
@@ -15,116 +16,202 @@ func UseS2STransfer(ctx *state.GenerateState) {
 	}
 
 	for fr, frNode := range ctx.FromNodes {
-		fromShard, ok := fr.(*ioswitch2.FromShardstore)
-		if !ok {
-			continue
+		switch fr := fr.(type) {
+		case *ioswitch2.FromShardstore:
+			s2sFromShardStore(ctx, fr, frNode)
+		case *ioswitch2.FromPublicStore:
+			s2sFromPublicStore(ctx, fr, frNode)
 		}
+	}
+}
 
-		fromStgBld := factory.GetBuilder(&fromShard.Space)
-		if !fromStgBld.FeatureDesc().HasBypassRead {
-			continue
-		}
+func s2sFromShardStore(ctx *state.GenerateState, fromShard *ioswitch2.FromShardstore, frNode ops2.FromNode) {
+	fromStgBld := factory.GetBuilder(&fromShard.UserSpace)
+	if !fromStgBld.FeatureDesc().HasBypassShardRead {
+		return
+	}
 
-		s2s, err := fromStgBld.CreateS2STransfer()
-		if err != nil {
-			continue
-		}
+	s2s, err := fromStgBld.CreateS2STransfer()
+	if err != nil {
+		return
+	}
 
-		// 此输出流的所有目的地都要能支持S2S传输
-		outVar := frNode.Output().Var()
-		if outVar.Dst.Len() == 0 {
-			continue
-		}
+	// 此输出流的所有目的地都要能支持S2S传输
+	outVar := frNode.Output().Var()
+	if outVar.Dst.Len() == 0 {
+		return
+	}
+
+	failed := false
+	var toShards []*ops2.ShardWriteNode
+	var toPublics []*ops2.PublicWriteNode
 
-		failed := false
-		var toShards []*ops2.ShardWriteNode
-		// var toShareds []*ops2.SharedLoadNode
-
-	loop:
-		for i := 0; i < outVar.Dst.Len(); i++ {
-			dstNode := outVar.Dst.Get(i)
-
-			switch dstNode := dstNode.(type) {
-			case *ops2.ShardWriteNode:
-				dstStgBld := factory.GetBuilder(&dstNode.UserSpace)
-				if !dstStgBld.FeatureDesc().HasBypassWrite {
-					failed = true
-					break
-				}
-
-				if !s2s.CanTransfer(&dstNode.UserSpace) {
-					failed = true
-					break
-				}
-
-				toShards = append(toShards, dstNode)
-
-				/* TODO 暂不支持共享存储服务
-				case *ops2.SharedLoadNode:
-					if !s2s.CanTransfer(to.Storage) {
-						failed = true
-						break
-					}
-					toShareds = append(toShareds, to)
-				*/
-			default:
+loop:
+	for i := 0; i < outVar.Dst.Len(); i++ {
+		dstNode := outVar.Dst.Get(i)
+
+		switch dstNode := dstNode.(type) {
+		case *ops2.ShardWriteNode:
+			dstStgBld := factory.GetBuilder(&dstNode.UserSpace)
+			if !dstStgBld.FeatureDesc().HasBypassShardWrite {
 				failed = true
-				break loop
+				break
 			}
-		}
-		if failed {
-			continue
-		}
 
-		for _, toShard := range toShards {
-			s2sNode := ctx.DAG.NewS2STransfer(fromShard.Space, toShard.UserSpace)
-			// 直传指令在目的地Hub上执行
-			s2sNode.Env().CopyFrom(toShard.Env())
+			if !s2s.CanTransfer(&dstNode.UserSpace) {
+				failed = true
+				break
+			}
+
+			toShards = append(toShards, dstNode)
 
-			// 先获取文件路径，送到S2S节点
-			brNode := ctx.DAG.NewBypassFromShardStore(fromShard.Space, fromShard.FileHash)
-			brNode.Env().CopyFrom(frNode.Env())
-			brNode.FilePathVar().ToSlot(s2sNode.SrcPathSlot())
+		case *ops2.PublicWriteNode:
+			dstStgBld := factory.GetBuilder(&dstNode.UserSpace)
+			if !dstStgBld.FeatureDesc().HasBypassPublicWrite {
+				failed = true
+				break
+			}
 
-			// 传输结果通知目的节点
-			bwNode := ctx.DAG.NewBypassToShardStore(toShard.UserSpace, toShard.To.FileHashStoreKey)
-			bwNode.Env().CopyFrom(toShard.Env())
+			if !s2s.CanTransfer(&dstNode.UserSpace) {
+				failed = true
+				break
+			}
 
-			s2sNode.BypassFileInfoVar().ToSlot(bwNode.BypassFileInfoSlot())
-			bwNode.BypassCallbackVar().ToSlot(s2sNode.BypassCallbackSlot())
+			toPublics = append(toPublics, dstNode)
 
-			// 从计划中删除目标节点
-			ctx.DAG.RemoveNode(toShard)
-			delete(ctx.ToNodes, toShard.To)
+		default:
+			failed = true
+			break loop
 		}
+	}
+	if failed {
+		return
+	}
+
+	for _, toShard := range toShards {
+		s2sNode := ctx.DAG.NewS2STransfer(fromShard.UserSpace, toShard.UserSpace, types.S2SOption{})
+		// 直传指令在目的地Hub上执行
+		s2sNode.Env().CopyFrom(toShard.Env())
+
+		// 先获取文件路径，送到S2S节点
+		brNode := ctx.DAG.NewBypassFromShardStore(fromShard.UserSpace, fromShard.FileHash)
+		brNode.Env().CopyFrom(frNode.Env())
+		brNode.FilePathVar().ToSlot(s2sNode.SrcPathSlot())
+
+		// 传输结果通知目的节点
+		bwNode := ctx.DAG.NewBypassToShardStore(toShard.UserSpace, toShard.To.FileHashStoreKey)
+		bwNode.Env().CopyFrom(toShard.Env())
+
+		s2sNode.BypassFileInfoVar().ToSlot(bwNode.BypassFileInfoSlot())
+		bwNode.BypassCallbackVar().ToSlot(s2sNode.BypassCallbackSlot())
+
+		// 从计划中删除目标节点
+		ctx.DAG.RemoveNode(toShard)
+		delete(ctx.ToNodes, toShard.To)
+	}
 
-		/*
-			for _, toShared := range toShareds {
-				s2sNode := ctx.DAG.NewS2STransfer(fromShard.Storage, toShared.Storage)
-				// 直传指令在目的地Hub上执行
-				s2sNode.Env().CopyFrom(toShared.Env())
+	for _, toPub := range toPublics {
+		s2sNode := ctx.DAG.NewS2STransfer(fromShard.UserSpace, toPub.UserSpace, types.S2SOption{
+			DestPathHint: toPub.ObjectPath,
+		})
+		// 直传指令在目的地Hub上执行
+		s2sNode.Env().CopyFrom(toPub.Env())
 
-				// 先获取文件路径，送到S2S节点
-				brNode := ctx.DAG.NewBypassFromShardStore(fromShard.Storage.Storage.StorageID, fromShard.FileHash)
-				brNode.Env().CopyFrom(toShared.Env())
-				brNode.FilePathVar().ToSlot(s2sNode.SrcPathSlot())
+		// 先获取文件路径，送到S2S节点
+		brNode := ctx.DAG.NewBypassFromShardStore(fromShard.UserSpace, fromShard.FileHash)
+		brNode.Env().CopyFrom(toPub.Env())
+		brNode.FilePathVar().ToSlot(s2sNode.SrcPathSlot())
 
-				// 传输结果通知目的节点
-				to := toShared.To.(*ioswitch2.LoadToShared)
-				bwNode := ctx.DAG.NewBypassToShardStore(toShard.Storage.Storage.StorageID, to.FileHashStoreKey)
-				bwNode.Env().CopyFrom(toShard.Env())
+		// 传输结果通知目的节点
+		bwNode := ctx.DAG.NewBypassToPublicStore(toPub.UserSpace, toPub.ObjectPath)
+		bwNode.Env().CopyFrom(toPub.Env())
 
-				s2sNode.BypassFileInfoVar().ToSlot(bwNode.BypassFileInfoSlot())
-				bwNode.BypassCallbackVar().ToSlot(s2sNode.BypassCallbackSlot())
+		s2sNode.BypassFileInfoVar().ToSlot(bwNode.BypassFileInfoSlot())
+		bwNode.BypassCallbackVar().ToSlot(s2sNode.BypassCallbackSlot())
 
-				// 从计划中删除目标节点
-				ctx.DAG.RemoveNode(toShared)
-				delete(ctx.ToNodes, toShared.To)
+		// 从计划中删除目标节点
+		ctx.DAG.RemoveNode(toPub)
+		delete(ctx.ToNodes, toPub.To)
+	}
+
+	// 从计划中删除源节点
+	ctx.DAG.RemoveNode(frNode)
+	delete(ctx.FromNodes, frNode.GetFrom())
+}
+
+func s2sFromPublicStore(ctx *state.GenerateState, fromPub *ioswitch2.FromPublicStore, frNode ops2.FromNode) {
+	fromStgBld := factory.GetBuilder(&fromPub.UserSpace)
+	if !fromStgBld.FeatureDesc().HasBypassPublicRead {
+		return
+	}
+
+	s2s, err := fromStgBld.CreateS2STransfer()
+	if err != nil {
+		return
+	}
+
+	// 此输出流的所有目的地都要能支持S2S传输
+	outVar := frNode.Output().Var()
+	if outVar.Dst.Len() == 0 {
+		return
+	}
+
+	failed := false
+	var toPublics []*ops2.PublicWriteNode
+
+loop:
+	for i := 0; i < outVar.Dst.Len(); i++ {
+		dstNode := outVar.Dst.Get(i)
+
+		switch dstNode := dstNode.(type) {
+		case *ops2.PublicWriteNode:
+			dstStgBld := factory.GetBuilder(&dstNode.UserSpace)
+			if !dstStgBld.FeatureDesc().HasBypassPublicWrite {
+				failed = true
+				break
+			}
+
+			if !s2s.CanTransfer(&dstNode.UserSpace) {
+				failed = true
+				break
 			}
-		*/
 
-		// 从计划中删除源节点
-		ctx.DAG.RemoveNode(frNode)
-		delete(ctx.FromNodes, fr)
+			toPublics = append(toPublics, dstNode)
+
+		default:
+			failed = true
+			break loop
+		}
+	}
+	if failed {
+		return
 	}
+
+	for _, toPub := range toPublics {
+		s2sNode := ctx.DAG.NewS2STransfer(fromPub.UserSpace, toPub.UserSpace, types.S2SOption{
+			DestPathHint: toPub.ObjectPath,
+		})
+		// 直传指令在目的地Hub上执行
+		s2sNode.Env().CopyFrom(toPub.Env())
+
+		// 先获取文件路径，送到S2S节点
+		brNode := ctx.DAG.NewBypassFromPublicStore(fromPub.UserSpace, fromPub.Path)
+		brNode.Env().CopyFrom(toPub.Env())
+		brNode.FilePathVar().ToSlot(s2sNode.SrcPathSlot())
+
+		// 传输结果通知目的节点
+		bwNode := ctx.DAG.NewBypassToPublicStore(toPub.UserSpace, toPub.ObjectPath)
+		bwNode.Env().CopyFrom(toPub.Env())
+
+		s2sNode.BypassFileInfoVar().ToSlot(bwNode.BypassFileInfoSlot())
+		bwNode.BypassCallbackVar().ToSlot(s2sNode.BypassCallbackSlot())
+
+		// 从计划中删除目标节点
+		ctx.DAG.RemoveNode(toPub)
+		delete(ctx.ToNodes, toPub.To)
+	}
+
+	// 从计划中删除源节点
+	ctx.DAG.RemoveNode(frNode)
+	delete(ctx.FromNodes, frNode.GetFrom())
 }

common/pkgs/storage/efile/ec_multiplier.go

@@ -24,7 +24,7 @@ type ECMultiplier struct {
 
 // 进行EC运算，coef * inputs。coef为编码矩阵，inputs为待编码数据，chunkSize为分块大小。
 // 输出为每一个块文件的路径，数组长度 = len(coef)
-func (m *ECMultiplier) Multiply(coef [][]byte, inputs []types.HTTPRequest, chunkSize int) ([]types.BypassUploadedFile, error) {
+func (m *ECMultiplier) Multiply(coef [][]byte, inputs []types.HTTPRequest, chunkSize int) ([]types.BypassedFileInfo, error) {
 	type Request struct {
 		Inputs    []types.HTTPRequest `json:"inputs"`
 		Outputs   []string            `json:"outputs"`
@@ -94,9 +94,9 @@ func (m *ECMultiplier) Multiply(coef [][]byte, inputs []types.HTTPRequest, chunk
 		return nil, fmt.Errorf("data length not match outputs length")
 	}
 
-	ret := make([]types.BypassUploadedFile, len(r.Data))
+	ret := make([]types.BypassedFileInfo, len(r.Data))
 	for i, data := range r.Data {
-		ret[i] = types.BypassUploadedFile{
+		ret[i] = types.BypassedFileInfo{
 			Path: m.outputs[i],
 			Size: data.Size,
 			Hash: clitypes.NewFullHashFromString(data.Sha256),

common/pkgs/storage/local/local.go

@@ -25,8 +25,8 @@ type builder struct {
 
 func (b *builder) FeatureDesc() types.FeatureDesc {
 	return types.FeatureDesc{
-		HasBypassWrite: true,
-		HasBypassRead:  true,
+		HasBypassShardWrite: true,
+		HasBypassShardRead:  true,
 	}
 }
 

common/pkgs/storage/local/multipart_upload.go

@@ -86,14 +86,14 @@ func (i *MultipartTask) InitState() types.MultipartInitState {
 	}
 }
 
-func (i *MultipartTask) JoinParts(ctx context.Context, parts []types.UploadedPartInfo) (types.BypassUploadedFile, error) {
+func (i *MultipartTask) JoinParts(ctx context.Context, parts []types.UploadedPartInfo) (types.BypassedFileInfo, error) {
 	parts = sort2.Sort(parts, func(l, r types.UploadedPartInfo) int {
 		return l.PartNumber - r.PartNumber
 	})
 
 	joined, err := os.Create(i.joinedFilePath)
 	if err != nil {
-		return types.BypassUploadedFile{}, err
+		return types.BypassedFileInfo{}, err
 	}
 	defer joined.Close()
 
@@ -103,14 +103,14 @@ func (i *MultipartTask) JoinParts(ctx context.Context, parts []types.UploadedPar
 	for _, part := range parts {
 		partSize, err := i.writePart(part, joined, hasher)
 		if err != nil {
-			return types.BypassUploadedFile{}, err
+			return types.BypassedFileInfo{}, err
 		}
 
 		size += partSize
 	}
 
 	h := hasher.Sum(nil)
-	return types.BypassUploadedFile{
+	return types.BypassedFileInfo{
 		Path: joined.Name(),
 		Size: size,
 		Hash: clitypes.NewFullHash(h),

common/pkgs/storage/local/s2s.go

@@ -9,6 +9,7 @@ import (
 
 	"gitlink.org.cn/cloudream/common/utils/os2"
 	clitypes "gitlink.org.cn/cloudream/jcs-pub/client/types"
+	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/types"
 	cortypes "gitlink.org.cn/cloudream/jcs-pub/coordinator/types"
 )
 
@@ -33,14 +34,16 @@ func (s *S2STransfer) CanTransfer(src *clitypes.UserSpaceDetail) bool {
 }
 
 // 执行数据直传
-func (s *S2STransfer) Transfer(ctx context.Context, src *clitypes.UserSpaceDetail, srcPath string) (string, error) {
+func (s *S2STransfer) Transfer(ctx context.Context, src *clitypes.UserSpaceDetail, srcPath string, opt types.S2SOption) (string, error) {
 	absTempDir, err := filepath.Abs(s.feat.TempDir)
 	if err != nil {
 		return "", fmt.Errorf("get abs temp dir %v: %v", s.feat.TempDir, err)
 	}
 
-	tempFileName := os2.GenerateRandomFileName(10)
-	s.dstPath = filepath.Join(absTempDir, tempFileName)
+	s.dstPath = opt.DestPathHint
+	if s.dstPath == "" {
+		s.dstPath = filepath.Join(absTempDir, os2.GenerateRandomFileName(10))
+	}
 
 	copy, err := os.OpenFile(s.dstPath, os.O_WRONLY|os.O_CREATE, 0644)
 	if err != nil {

common/pkgs/storage/local/shard_store.go

@@ -399,9 +399,9 @@ func (s *ShardStore) getFilePathFromHash(hash clitypes.FileHash) string {
 	return filepath.Join(s.absRoot, BlocksDir, hash.GetHashPrefix(2), string(hash))
 }
 
-var _ types.BypassWrite = (*ShardStore)(nil)
+var _ types.BypassShardWrite = (*ShardStore)(nil)
 
-func (s *ShardStore) BypassUploaded(info types.BypassUploadedFile) error {
+func (s *ShardStore) BypassedShard(info types.BypassedFileInfo) error {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
@@ -433,9 +433,9 @@ func (s *ShardStore) BypassUploaded(info types.BypassUploadedFile) error {
 	return nil
 }
 
-var _ types.BypassRead = (*ShardStore)(nil)
+var _ types.BypassShardRead = (*ShardStore)(nil)
 
-func (s *ShardStore) BypassRead(fileHash clitypes.FileHash) (types.BypassFilePath, error) {
+func (s *ShardStore) BypassShardRead(fileHash clitypes.FileHash) (types.BypassFilePath, error) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 

common/pkgs/storage/obs/obs.go

@@ -31,9 +31,11 @@ func newBuilder(detail *clitypes.UserSpaceDetail) types.StorageBuilder {
 
 func (b *builder) FeatureDesc() types.FeatureDesc {
 	return types.FeatureDesc{
-		HasBypassWrite:    true,
-		HasBypassRead:     true,
-		HasBypassHTTPRead: true,
+		HasBypassShardWrite:  true,
+		HasBypassPublicWrite: true,
+		HasBypassShardRead:   true,
+		HasBypassPublicRead:  true,
+		HasBypassHTTPRead:    true,
 	}
 }
 

common/pkgs/storage/obs/obs_test.go

@@ -6,6 +6,7 @@ import (
 
 	. "github.com/smartystreets/goconvey/convey"
 	clitypes "gitlink.org.cn/cloudream/jcs-pub/client/types"
+	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/types"
 	cortypes "gitlink.org.cn/cloudream/jcs-pub/coordinator/types"
 )
 
@@ -39,7 +40,7 @@ func Test_S2S(t *testing.T) {
 			Storage: cortypes.Storage{
 				Type: &cortypes.OBSType{},
 			},
-		}, "test_data/test03.txt")
+		}, "test_data/test03.txt", types.S2SOption{})
 		defer s2s.Abort()
 
 		So(err, ShouldEqual, nil)

common/pkgs/storage/obs/s2s.go

@@ -13,6 +13,7 @@ import (
 	"gitlink.org.cn/cloudream/common/utils/os2"
 	clitypes "gitlink.org.cn/cloudream/jcs-pub/client/types"
 	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/s3"
+	"gitlink.org.cn/cloudream/jcs-pub/common/pkgs/storage/types"
 	cortypes "gitlink.org.cn/cloudream/jcs-pub/coordinator/types"
 )
 
@@ -37,7 +38,7 @@ func (s *S2STransfer) CanTransfer(src *clitypes.UserSpaceDetail) bool {
 }
 
 // 执行数据直传。返回传输后的文件路径
-func (s *S2STransfer) Transfer(ctx context.Context, src *clitypes.UserSpaceDetail, srcPath string) (string, error) {
+func (s *S2STransfer) Transfer(ctx context.Context, src *clitypes.UserSpaceDetail, srcPath string, opt types.S2SOption) (string, error) {
 	req := s.makeRequest(src, srcPath)
 	if req == nil {
 		return "", fmt.Errorf("unsupported source storage type: %T", src.Storage.Type)

common/pkgs/storage/s3/multipart_upload.go

@@ -101,7 +101,7 @@ func (i *MultipartTask) InitState() types.MultipartInitState {
 	}
 }
 
-func (i *MultipartTask) JoinParts(ctx context.Context, parts []types.UploadedPartInfo) (types.BypassUploadedFile, error) {
+func (i *MultipartTask) JoinParts(ctx context.Context, parts []types.UploadedPartInfo) (types.BypassedFileInfo, error) {
 	parts = sort2.Sort(parts, func(l, r types.UploadedPartInfo) int {
 		return l.PartNumber - r.PartNumber
 	})
@@ -127,7 +127,7 @@ func (i *MultipartTask) JoinParts(ctx context.Context, parts []types.UploadedPar
 		},
 	})
 	if err != nil {
-		return types.BypassUploadedFile{}, err
+		return types.BypassedFileInfo{}, err
 	}
 
 	headResp, err := i.cli.HeadObject(ctx, &s3.HeadObjectInput{
@@ -135,12 +135,12 @@ func (i *MultipartTask) JoinParts(ctx context.Context, parts []types.UploadedPar
 		Key:    aws.String(i.tempFilePath),
 	})
 	if err != nil {
-		return types.BypassUploadedFile{}, err
+		return types.BypassedFileInfo{}, err
 	}
 
 	hash := clitypes.CalculateCompositeHash(partHashes)
 
-	return types.BypassUploadedFile{
+	return types.BypassedFileInfo{
 		Path: i.tempFilePath,
 		Size: *headResp.ContentLength,
 		Hash: hash,

common/pkgs/storage/s3/public_store.go

@@ -103,3 +103,40 @@ func (s *PublicStore) ListAll(path string) ([]types.PublicStoreEntry, error) {
 func (s *PublicStore) getLogger() logger.Logger {
 	return logger.WithField("PublicStore", "S3").WithField("Storage", s.Detail.Storage.String())
 }
+
+var _ types.BypassPublicRead = (*PublicStore)(nil)
+
+func (s *PublicStore) BypassPublicRead(pa string) (types.BypassFilePath, error) {
+	info, err := s.cli.HeadObject(context.TODO(), &s3.HeadObjectInput{
+		Bucket: aws.String(s.Bucket),
+		Key:    aws.String(pa),
+	})
+	if err != nil {
+		s.getLogger().Warnf("get file %v: %v", pa, err)
+		return types.BypassFilePath{}, err
+	}
+
+	return types.BypassFilePath{
+		Path: pa,
+		Info: types.FileInfo{
+			Size:        *info.ContentLength,
+			Description: pa,
+		},
+	}, nil
+}
+
+var _ types.BypassPublicWrite = (*PublicStore)(nil)
+
+func (s *PublicStore) BypassedPublic(info types.BypassedFileInfo, dstPath string) error {
+	_, err := s.cli.CopyObject(context.TODO(), &s3.CopyObjectInput{
+		Bucket:     aws.String(s.Bucket),
+		CopySource: aws.String(s.Bucket + "/" + info.Path),
+		Key:        aws.String(dstPath),
+	})
+	if err != nil {
+		s.getLogger().Warnf("bypass public write %v to %v: %v", info.Path, dstPath, err)
+		return err
+	}
+
+	return nil
+}

common/pkgs/storage/s3/s3.go

@@ -30,9 +30,11 @@ func newBuilder(detail *clitypes.UserSpaceDetail) types.StorageBuilder {
 
 func (b *builder) FeatureDesc() types.FeatureDesc {
 	return types.FeatureDesc{
-		HasBypassWrite:    true,
-		HasBypassRead:     true,
-		HasBypassHTTPRead: false,
+		HasBypassShardWrite:  true,
+		HasBypassPublicWrite: true,
+		HasBypassShardRead:   true,
+		HasBypassPublicRead:  true,
+		HasBypassHTTPRead:    false,
 	}
 }
 

common/pkgs/storage/s3/shard_store.go

@@ -461,9 +461,9 @@ func (s *ShardStore) GetFilePathFromHash(hash clitypes.FileHash) string {
 	return JoinKey(s.Detail.UserSpace.ShardStore.BaseDir, BlocksDir, hash.GetHashPrefix(2), string(hash))
 }
 
-var _ types.BypassWrite = (*ShardStore)(nil)
+var _ types.BypassShardWrite = (*ShardStore)(nil)
 
-func (s *ShardStore) BypassUploaded(info types.BypassUploadedFile) error {
+func (s *ShardStore) BypassedShard(info types.BypassedFileInfo) error {
 	if info.Hash == "" {
 		return fmt.Errorf("empty file hash is not allowed by this shard store")
 	}
@@ -498,9 +498,9 @@ func (s *ShardStore) BypassUploaded(info types.BypassUploadedFile) error {
 	return nil
 }
 
-var _ types.BypassRead = (*ShardStore)(nil)
+var _ types.BypassShardRead = (*ShardStore)(nil)
 
-func (s *ShardStore) BypassRead(fileHash clitypes.FileHash) (types.BypassFilePath, error) {
+func (s *ShardStore) BypassShardRead(fileHash clitypes.FileHash) (types.BypassFilePath, error) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 

common/pkgs/storage/types/bypass.go

@@ -5,33 +5,42 @@ import (
 )
 
 // 通过旁路上传后的文件的信息
-type BypassUploadedFile struct {
+type BypassedFileInfo struct {
 	Path string
 	Hash clitypes.FileHash
 	Size int64
 }
 
 // 不通过ShardStore上传文件，但上传完成后需要通知ShardStore。
-// 也可以用于共享存储。
-type BypassWrite interface {
-	BypassUploaded(info BypassUploadedFile) error
+type BypassShardWrite interface {
+	BypassedShard(info BypassedFileInfo) error
+}
+
+// 不通过PublicStore上传文件。
+type BypassPublicWrite interface {
+	BypassedPublic(info BypassedFileInfo, dstPath string) error
 }
 
 // 描述指定文件在分片存储中的路径。可以考虑设计成interface。
 type BypassFilePath struct {
 	Path string
+	// 文件信息。此结构体中的字段不一定都会存在，比如如果是从BypassPublicRead接口获取的，那么Hash字段就为空。
 	Info FileInfo
 }
 
 // 不通过ShardStore读取文件，但需要它返回文件的路径。
-// 仅用于分片存储。
-type BypassRead interface {
-	BypassRead(fileHash clitypes.FileHash) (BypassFilePath, error)
+type BypassShardRead interface {
+	BypassShardRead(fileHash clitypes.FileHash) (BypassFilePath, error)
+}
+
+// 不通过PublicStore读取文件。虽然仅使用path就已经可以读取到文件，但还是增加了此接口用于获取更详细的文件信息。
+type BypassPublicRead interface {
+	BypassPublicRead(path string) (BypassFilePath, error)
 }
 
 // 能通过一个Http请求直接访问文件
 // 仅用于分片存储。
-type HTTPBypassRead interface {
+type HTTPBypassShardRead interface {
 	HTTPBypassRead(fileHash clitypes.FileHash) (HTTPRequest, error)
 }
 

common/pkgs/storage/types/ec_multiplier.go

@@ -3,7 +3,7 @@ package types
 type ECMultiplier interface {
 	// 进行EC运算，coef * inputs。coef为编码矩阵，inputs为待编码数据，chunkSize为分块大小。
 	// 输出为每一个块文件的路径，数组长度 = len(coef)
-	Multiply(coef [][]byte, inputs []HTTPRequest, chunkSize int) ([]BypassUploadedFile, error)
+	Multiply(coef [][]byte, inputs []HTTPRequest, chunkSize int) ([]BypassedFileInfo, error)
 	// 完成计算
 	Complete()
 	// 取消计算。如果已经调用了Complete，则应该无任何影响

common/pkgs/storage/types/s2s.go

@@ -6,11 +6,17 @@ import (
 	clitypes "gitlink.org.cn/cloudream/jcs-pub/client/types"
 )
 
+type S2SOption struct {
+	// 传输目的地路径。如果为空，则自动生成一个临时路径。
+	// 只是一个建议，非强制，如果存储服务不支持直接指定目的路径，则可以忽略这个参数。
+	DestPathHint string
+}
+
 type S2STransfer interface {
 	// 判断是否能从指定的源存储中直传到当前存储的目的路径。仅在生成计划时使用
 	CanTransfer(src *clitypes.UserSpaceDetail) bool
 	// 执行数据直传。返回传输后的文件路径
-	Transfer(ctx context.Context, src *clitypes.UserSpaceDetail, srcPath string) (string, error)
+	Transfer(ctx context.Context, src *clitypes.UserSpaceDetail, srcPath string, opt S2SOption) (string, error)
 	// 完成传输
 	Complete()
 	// 取消传输。如果已经调用了Complete，则这个方法应该无效果

common/pkgs/storage/types/s3_client.go

@@ -17,7 +17,7 @@ type Multiparter interface {
 type MultipartTask interface {
 	InitState() MultipartInitState
 	// 所有分片上传完成后，合并分片
-	JoinParts(ctx context.Context, parts []UploadedPartInfo) (BypassUploadedFile, error)
+	JoinParts(ctx context.Context, parts []UploadedPartInfo) (BypassedFileInfo, error)
 	// 合成之后的文件已被使用
 	Complete()
 	// 取消上传。如果在调用Complete之前调用，则应该删除合并后的文件。如果已经调用Complete，则应该不做任何事情。

common/pkgs/storage/types/types.go

@@ -32,10 +32,14 @@ type StorageBuilder interface {
 }
 
 type FeatureDesc struct {
-	// 是否能旁路上传
-	HasBypassWrite bool
-	// 是否能旁路读取
-	HasBypassRead bool
+	// 是否能旁路上传分片
+	HasBypassShardWrite bool
+	// 是否能旁路上传公共存储
+	HasBypassPublicWrite bool
+	// 是否能旁路读取分片
+	HasBypassShardRead bool
+	// 公共存储是否支持旁路读取
+	HasBypassPublicRead bool
 	// 是否能通过HTTP读取
 	HasBypassHTTPRead bool
 }

go.mod

@@ -73,7 +73,6 @@ require (
 	golang.org/x/exp v0.0.0-20250218142911-aa4b98e5adaa // indirect
 	golang.org/x/net v0.35.0 // indirect
 	golang.org/x/text v0.24.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20241015192408-796eee8c2d53 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20241021214115-324edc3d5d38 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
@@ -82,14 +81,10 @@ require (
 require (
 	github.com/antonfisher/nested-logrus-formatter v1.3.1 // indirect
 	github.com/c-bata/go-prompt v0.2.6
-	github.com/coreos/go-semver v0.3.1 // indirect
-	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
 	github.com/gin-contrib/sse v0.1.0 // indirect
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-playground/validator/v10 v10.20.0 // indirect
-	github.com/gogo/protobuf v1.3.2 // indirect
-	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/gopherjs/gopherjs v1.17.2 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
@@ -111,10 +106,5 @@ require (
 	github.com/streadway/amqp v1.1.0
 	github.com/ugorji/go/codec v1.2.12 // indirect
 	github.com/zyedidia/generic v1.2.1 // indirect
-	go.etcd.io/etcd/api/v3 v3.5.12 // indirect
-	go.etcd.io/etcd/client/pkg/v3 v3.5.12 // indirect
-	go.etcd.io/etcd/client/v3 v3.5.12 // indirect
-	go.uber.org/multierr v1.10.0 // indirect
-	go.uber.org/zap v1.27.0 // indirect
 	gorm.io/driver/mysql v1.5.7
 )

go.sum

@@ -41,10 +41,6 @@ github.com/cloudwego/base64x v0.1.4/go.mod h1:0zlkT4Wn5C6NdauXdJRhSKRlJvmclQ1hhJ
 github.com/cloudwego/iasm v0.2.0 h1:1KNIy1I1H9hNNFEEH3DVnI4UujN+1zjpuk6gwHLTssg=
 github.com/cloudwego/iasm v0.2.0/go.mod h1:8rXZaNYT2n95jn+zTI1sDr+IgcD2GVs0nlbbQPiEFhY=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
-github.com/coreos/go-semver v0.3.1 h1:yi21YpKnrx1gt5R+la8n5WgS0kCrsPp33dmEyHReZr4=
-github.com/coreos/go-semver v0.3.1/go.mod h1:irMmmIw/7yzSRPWryHsK7EYSg09caPQL03VsM8rvUec=
-github.com/coreos/go-systemd/v22 v22.5.0 h1:RrqgGjYQKalulkV8NGVIfkXQf6YYmOyiJKk8iXXhfZs=
-github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
 github.com/cpuguy83/go-md2man/v2 v2.0.4/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -75,9 +71,6 @@ github.com/go-sql-driver/mysql v1.8.1 h1:LedoTUt/eveggdHS9qUFC1EFSa8bU2+1pZjSRpv
 github.com/go-sql-driver/mysql v1.8.1/go.mod h1:wEBSXgmK//2ZFJyE+qWnIsVGmvmEKlqwuVSjsCm7DZg=
 github.com/goccy/go-json v0.10.3 h1:KZ5WoDbxAIgm2HNbYckL0se1fHD6rz5j4ywS6ebzDqA=
 github.com/goccy/go-json v0.10.3/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
-github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
-github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
-github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
 github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
@@ -89,8 +82,6 @@ github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrU
 github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:WU3c8KckQ9AFe+yFwt9sWVRKCVIyN9cPHBJSNnbL67w=
 github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0=
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
-github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
-github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
@@ -137,8 +128,6 @@ github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnr
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
 github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
-github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
-github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.2.8 h1:+StwCXwm9PdpiEkPyzBXIy+M9KUb4ODm0Zarf1kS5BM=
@@ -237,30 +226,17 @@ github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT
 github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4=
 github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
 github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7JulP+udvsHwJoVG1YGAP6VLg4y9I5dyZdqmA=
-github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
-github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 github.com/zyedidia/generic v1.2.1 h1:Zv5KS/N2m0XZZiuLS82qheRG4X1o5gsWreGb0hR7XDc=
 github.com/zyedidia/generic v1.2.1/go.mod h1:ly2RBz4mnz1yeuVbQA/VFwGjK3mnHGRj1JuoG336Bis=
-go.etcd.io/etcd/api/v3 v3.5.12 h1:W4sw5ZoU2Juc9gBWuLk5U6fHfNVyY1WC5g9uiXZio/c=
-go.etcd.io/etcd/api/v3 v3.5.12/go.mod h1:Ot+o0SWSyT6uHhA56al1oCED0JImsRiU9Dc26+C2a+4=
-go.etcd.io/etcd/client/pkg/v3 v3.5.12 h1:EYDL6pWwyOsylrQyLp2w+HkQ46ATiOvoEdMarindU2A=
-go.etcd.io/etcd/client/pkg/v3 v3.5.12/go.mod h1:seTzl2d9APP8R5Y2hFL3NVlD6qC/dOT+3kvrqPyTas4=
-go.etcd.io/etcd/client/v3 v3.5.12 h1:v5lCPXn1pf1Uu3M4laUE2hp/geOTc5uPcYYsNe1lDxg=
-go.etcd.io/etcd/client/v3 v3.5.12/go.mod h1:tSbBCakoWmmddL+BKVAJHa9km+O/E+bumDe9mSbPiqw=
 go.mongodb.org/mongo-driver v1.12.0 h1:aPx33jmn/rQuJXPQLZQ8NtfPQG8CaqgLThFtqRb0PiE=
 go.mongodb.org/mongo-driver v1.12.0/go.mod h1:AZkxhPnFJUoH7kZlFkVKucV20K387miPfm7oimrSmK0=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
-go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ=
-go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
-go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
-go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
 golang.org/x/arch v0.0.0-20210923205945-b76863e36670/go.mod h1:5om86z9Hs0C8fWVUuoMHwpExlXzs5Tkyp9hOrfG7pp8=
 golang.org/x/arch v0.8.0 h1:3wRIsP3pM4yUptoR96otTUOXI367OS0+c9eeRi9doIc=
 golang.org/x/arch v0.8.0/go.mod h1:FEVrYAQjsQXMVJ1nsMoVVXPZg6p2JE2mx8psSWTDQys=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
-golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20201012173705-84dcc777aaee/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
@@ -275,8 +251,6 @@ golang.org/x/exp v0.0.0-20250218142911-aa4b98e5adaa/go.mod h1:BHOTPb3L19zxehTsLo
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
 golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
-golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
-golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
 golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -285,9 +259,7 @@ golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73r
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20201010224723-4f7140c49acb/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
@@ -301,7 +273,6 @@ golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.13.0 h1:AauUjRAJ9OSnvULf/ARrrVywoJDy0YS2AwQ98I37610=
@@ -356,20 +327,14 @@ golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
-golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
-golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
 golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
 google.golang.org/genproto v0.0.0-20190819201941-24fa4b261c55/go.mod h1:DMBHOl98Agz4BDEuKkezgsaosCRResVns1a3J2ZsMNc=
-google.golang.org/genproto/googleapis/api v0.0.0-20241015192408-796eee8c2d53 h1:fVoAXEKA4+yufmbdVYv+SE73+cPZbbbe8paLsHfkK+U=
-google.golang.org/genproto/googleapis/api v0.0.0-20241015192408-796eee8c2d53/go.mod h1:riSXTwQ4+nqmPGtobMFyW5FqVAmIs0St6VPp4Ug7CE4=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20241021214115-324edc3d5d38 h1:zciRKQ4kBpFgpfC5QQCVtnnNAcLIqweL7plyZRQHVpI=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20241021214115-324edc3d5d38/go.mod h1:GX3210XPVPUjJbTUbvwI8f2IpZDMZuPJWDzDuebbviI=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=