Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
services: add new service for fetching blocks from NeoFS
Close #3496 Signed-off-by: Ekaterina Pavlova <[email protected]>
- Loading branch information
1 parent
79e7898
commit 616393d
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| package config | ||
|
|
||
| import "time" | ||
|
|
||
| // NeoFS represents the configuration for the blockfetcher service. | ||
| type ( | ||
| NeoFS struct { | ||
| Nodes []string `yaml:"Nodes"` | ||
| Timeout time.Duration `yaml:"Timeout"` | ||
| ContainerID string `yaml:"ContainerID"` | ||
| DumpDir string `yaml:"DumpDir"` | ||
| Restore bool `yaml:"Restore"` | ||
| } | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,214 @@ | ||
| package blockfetcher | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "fmt" | ||
| "io" | ||
| "net/url" | ||
| "time" | ||
|
|
||
| "github.com/nspcc-dev/neo-go/pkg/config" | ||
| "github.com/nspcc-dev/neo-go/pkg/core/block" | ||
| "github.com/nspcc-dev/neo-go/pkg/core/chaindump" | ||
| "github.com/nspcc-dev/neo-go/pkg/core/storage" | ||
| "github.com/nspcc-dev/neo-go/pkg/crypto/keys" | ||
| gio "github.com/nspcc-dev/neo-go/pkg/io" | ||
| "github.com/nspcc-dev/neo-go/pkg/services/oracle/neofs" | ||
| "github.com/nspcc-dev/neo-go/pkg/util" | ||
| "github.com/nspcc-dev/neofs-sdk-go/client" | ||
| cid "github.com/nspcc-dev/neofs-sdk-go/container/id" | ||
| "github.com/nspcc-dev/neofs-sdk-go/object" | ||
| oid "github.com/nspcc-dev/neofs-sdk-go/object/id" | ||
| "go.uber.org/zap" | ||
| ) | ||
|
|
||
| // Ledger is an interface to Blockchain sufficient for Service. | ||
| type Ledger interface { | ||
| LastBatch() *storage.MemBatch | ||
| AddBlock(block *block.Block) error | ||
| GetBlock(hash util.Uint256) (*block.Block, error) | ||
| GetConfig() config.Blockchain | ||
| GetHeaderHash(u uint32) util.Uint256 | ||
| } | ||
|
|
||
| type Service struct { | ||
| chain Ledger | ||
| log *zap.Logger | ||
| client *client.Client | ||
| quit chan bool | ||
| containerID cid.ID | ||
| Timeout time.Duration | ||
| Nodes []string | ||
| dumpDir string | ||
| } | ||
|
|
||
| // ConfigObject represents the configuration object in NeoFS. | ||
| type ConfigObject struct { | ||
| HashOIDs []string `json:"hash_oid"` | ||
| BlockOIDs []string `json:"block_oid"` | ||
| Height uint32 `json:"height"` | ||
| Timestamp int64 `json:"timestamp"` | ||
| Step uint32 `json:"step"` | ||
| } | ||
|
|
||
| // New creates a new BlockFetcherService. | ||
| func New(chain Ledger, cfg config.NeoFS, logger *zap.Logger) *Service { | ||
| neofsClient, err := client.New(client.PrmInit{}) | ||
| if err != nil { | ||
| logger.Error("Failed to create NeoFS client", zap.Error(err)) | ||
| return nil | ||
| } | ||
| var containerID cid.ID | ||
| err = containerID.DecodeString(cfg.ContainerID) | ||
| if err != nil { | ||
| logger.Error("Failed to decode container ID", zap.Error(err)) | ||
| return nil | ||
| } | ||
| return &Service{ | ||
| chain: chain, | ||
| log: logger, | ||
| client: neofsClient, | ||
| quit: make(chan bool), | ||
| dumpDir: cfg.DumpDir, | ||
| containerID: containerID, | ||
| Nodes: cfg.Nodes, | ||
| Timeout: cfg.Timeout, | ||
| } | ||
| } | ||
|
|
||
| // Name implements the core.Service interface. | ||
| func (bfs *Service) Name() string { | ||
| return "BlockFetcherService" | ||
| } | ||
|
|
||
| // Start implements the core.Service interface. | ||
| func (bfs *Service) Start(done func()) { | ||
| bfs.log.Info("Starting Block Fetcher Service") | ||
| go func() { | ||
| err := bfs.fetchData() | ||
| if err != nil { | ||
| close(bfs.quit) | ||
| return | ||
| } | ||
| done() | ||
| }() | ||
| } | ||
|
|
||
| // Shutdown implements the core.Service interface. | ||
| func (bfs *Service) Shutdown() { | ||
| bfs.log.Info("Shutting down Block Fetcher Service") | ||
| close(bfs.quit) | ||
| } | ||
|
|
||
| func (bfs *Service) fetchData() error { | ||
| select { | ||
| case <-bfs.quit: | ||
| return nil | ||
| default: | ||
| prm := client.PrmObjectSearch{} | ||
| filters := object.NewSearchFilters() | ||
| filters.AddFilter("type", "config", object.MatchStringEqual) | ||
| prm.SetFilters(filters) | ||
|
|
||
| configOid, err := bfs.search(prm) | ||
| if err != nil { | ||
| bfs.log.Error("Failed to fetch object IDs", zap.Error(err)) | ||
| return err | ||
| } | ||
| cfg, err := bfs.get(configOid[0].String()) | ||
| if err != nil { | ||
| bfs.log.Error("Failed to fetch config object", zap.Error(err)) | ||
| return err | ||
| } | ||
| var configObj ConfigObject | ||
| err = json.Unmarshal(cfg, &configObj) | ||
| if err != nil { | ||
| bfs.log.Error("Failed to unmarshal configuration data", zap.Error(err)) | ||
| return err | ||
| } | ||
|
|
||
| for _, HashOID := range configObj.HashOIDs { | ||
| _, err = bfs.get(HashOID) | ||
| if err != nil { | ||
| bfs.log.Error("Failed to fetch hash", zap.Error(err)) | ||
| return err | ||
| } | ||
| } | ||
|
|
||
| for _, blockOID := range configObj.BlockOIDs { | ||
| data, err := bfs.get(blockOID) | ||
| if err != nil { | ||
| bfs.log.Error(fmt.Sprintf("Failed to fetch block %s", blockOID), zap.Error(err)) | ||
| return err | ||
| } | ||
| err = bfs.ProcessBlock(data, configObj.Step) | ||
| if err != nil { | ||
| bfs.log.Error(fmt.Sprintf("Failed to process block %s", blockOID), zap.Error(err)) | ||
| return err | ||
| } | ||
| } | ||
| } | ||
| close(bfs.quit) | ||
| return nil | ||
| } | ||
|
|
||
| func (bfs *Service) ProcessBlock(data []byte, count uint32) error { | ||
| br := gio.NewBinReaderFromBuf(data) | ||
| dump := NewDump() | ||
| var lastIndex uint32 | ||
|
|
||
| err := chaindump.Restore(bfs.chain, br, 0, count, func(b *block.Block) error { | ||
| batch := bfs.chain.LastBatch() | ||
| if batch != nil { | ||
| dump.Add(b.Index, batch) | ||
| lastIndex = b.Index | ||
| if b.Index%1000 == 0 { | ||
| if err := dump.TryPersist(bfs.dumpDir, lastIndex); err != nil { | ||
| return fmt.Errorf("can't dump storage to file: %w", err) | ||
| } | ||
| } | ||
| } | ||
| return nil | ||
| }) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to restore blocks: %w", err) | ||
| } | ||
|
|
||
| if err = dump.TryPersist(bfs.dumpDir, lastIndex); err != nil { | ||
| return fmt.Errorf("final persistence failed: %w", err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (bfs *Service) get(oid string) ([]byte, error) { | ||
| privateKey, err := keys.NewPrivateKey() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| ctx, cancel := context.WithTimeout(context.Background(), bfs.Timeout) | ||
| defer cancel() | ||
| u, err := url.Parse(fmt.Sprintf("neofs:%s/%s", bfs.containerID, oid)) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| rc, err := neofs.GetWithClient(ctx, bfs.client, privateKey, u, bfs.Nodes[0]) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| data, err := io.ReadAll(rc) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return data, nil | ||
| } | ||
|
|
||
| func (bfs *Service) search(prm client.PrmObjectSearch) ([]oid.ID, error) { | ||
| privateKey, err := keys.NewPrivateKey() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| ctx, cancel := context.WithTimeout(context.Background(), bfs.Timeout) | ||
| defer cancel() | ||
| return neofs.ObjectSearch(ctx, bfs.client, privateKey, bfs.containerID, bfs.Nodes[0], prm) | ||
| } | ||
Oops, something went wrong.