Function EsNodeSync

cmd/es-node/main.go:305–356

(ctx *cli.Context)

}

func EsNodeSync(ctx *cli.Context) error {
	lg.Info("Sync data for specified kv")
	if !ctx.IsSet(kvIndexFlagName) {
		return fmt.Errorf("kv_index must be specified")
	}
	kvIndex := uint64(ctx.Int(kvIndexFlagName))
	lg.Info("Read flag", "name", kvIndexFlagName, "value", kvIndex)
	if !ctx.IsSet(esRpcFlagName) {
		return fmt.Errorf("es_rpc must be specified")
	}
	esRpc := ctx.String(esRpcFlagName)
	lg.Info("Read flag", "name", esRpcFlagName, "value", esRpc)
	// query meta
	contract := readRequiredFlag(ctx, flags.StorageL1Contract)
	if !common.IsHexAddress(contract) {
		return fmt.Errorf("invalid contract address %s", contract)
	}
	l1contract := common.HexToAddress(contract)
	l1Rpc := readRequiredFlag(ctx, flags.L1NodeAddr)
	pClient, err := eth.Dial(l1Rpc, l1contract, 2, lg)
	if err != nil {
		return fmt.Errorf("failed to dial eth client: %w", err)
	}
	meta, err := pClient.GetKvMetas([]uint64{kvIndex}, rpc.LatestBlockNumber.Int64())
	if err != nil {
		return fmt.Errorf("failed to get meta: %w", err)
	}
	lg.Info("Query meta from contract done", "kvIndex", kvIndex, "meta", common.Hash(meta[0]).Hex())
	// query blob
	var commit common.Hash
	copy(commit[:], meta[0][32-ethstorage.HashSizeInContract:32])
	blob, err := downloadBlobFromRPC(esRpc, kvIndex, commit)
	if err != nil {
		return fmt.Errorf("failed to download blob from RPC: %w", err)
	}
	lg.Info("Download blob from RPC done", "kvIndex", kvIndex, "commit", commit.Hex())
	// write blob and meta
	shardManager, err := initShardManager(ctx, l1Rpc, l1contract)
	if err != nil {
		return fmt.Errorf("failed to init shard manager: %w", err)
	}
	preparedCommit := ethstorage.PrepareCommit(commit)
	ok, err := shardManager.TryWrite(kvIndex, blob, preparedCommit)
	if err != nil {
		return fmt.Errorf("failed to write kv: %w", err)
	}
	if !ok {
		return fmt.Errorf("failed to write kv: kv index not in the shard")
	}
	lg.Info("Sync data finished", "kvIndex", kvIndex, "commit", commit.Hex())
	return nil
}