$ juicefs dump tikv://ip:2379/foo-dev foo-dev-dump.json
<INFO>: Meta address: tikv://<ip>:2379/foo-dev [interface.go:406]
<WARNING>: Secret key is removed for the sake of safety [tkv.go:2571]
           Scan keys count: 357806 / 357806 [===========================]  done
      Dumped entries count: 122527 / 122527 [===========================]  done
<INFO>: Dump metadata into dump succeed [dump.go:76]

{  "Setting": {
    "Name": "foo-dev",
    "UUID": "ca95c258",
    "Storage": "OSS",
    "Bucket": "http://<url>",
    "AccessKey": "ak",
    "BlockSize": 4096,
    "Capacity": 0,
    "Inodes": 0,
    "MetaVersion": 0,
    "MinClientVersion": "",
    "MaxClientVersion": "",
  },
  "Counters": {
    "usedSpace": 6164512768,
    "usedInodes": 5010,
    "nextInodes": 10402,
    "nextChunk": 25001,
    "nextSession": 118,
  },
  "FSTree": {
    "attr": {"inode":1,"type":"directory","mode":511,"atime":1645791488,"mtime":1652433235,"ctime":1652433235,"mtimensec":553010494,"ctimensec":553010494,"nlink":2,"length":0},
    "xattrs": [{"name":"lastBackup","value":"2024-05-30T13:50:25+08:00"}],
    "entries": {
      "001eb8b": {
        "attr": {...},
        "chunks": [{"index":0,"slices":[{"chunkid":15931,"size":32,"len":32}]}]
        ...
      }
   }
  },
}

$ juicefs load tikv://<ip>:2379/foo-dev-new foo-dev-dump.json

// tidb br/pkg/task/backup_txn.go

// RunBackupTxn starts a backup task inside the current goroutine.
func RunBackupTxn(c context.Context, g glue.Glue, cmdName string, cfg *TxnKvConfig) error {
    mgr := NewMgr(ctx, g, cfg.PD, cfg.TLS, GetKeepalive(&cfg.Config), cfg.CheckRequirements, false)
    client := backup.NewBackupClient(ctx, mgr)

    backupRanges := make([]rtree.Range, 0, 1)

    // current just build full txn range to support full txn backup
    minStartKey := []byte{}
    maxEndKey := []byte{}
    backupRanges = append(backupRanges, rtree.Range{
        StartKey: minStartKey,
        EndKey:   maxEndKey,
    })

    // Backup
    req := backuppb.BackupRequest{
        ClusterId:        client.GetClusterID(),
        StartVersion:     0,
        EndVersion:       client.GetCurrentTS(ctx), // gets a new timestamp (TSO) from PD
        RateLimit:        cfg.RateLimit,
        Concurrency:      cfg.Concurrency,
        StorageBackend:   client.GetStorageBackend(),
        IsRawKv:          false,
    }

    ranges, schemas, policies := client.BuildBackupRangeAndSchema(mgr.GetStorage(), cfg.TableFilter, backupTS, isFullBackup(cmdName))

    // StartWriteMetasAsync writes four kind of meta into backupmeta.
    // 1. file
    // 2. schema
    // 3. ddl
    // 4. rawRange( raw kv )
    metaWriter := metautil.NewMetaWriter(client.GetStorage(), metautil.MetaFileSize, false, metautil.MetaFile, &cfg.CipherInfo)
    metaWriter.StartWriteMetasAsync(ctx, metautil.AppendDataFile)

    // Start TiKV backup
    client.BackupRanges(ctx, backupRanges, req, 1, nil, metaWriter, progressCallBack)

    // Backup has finished
    metaWriter.Update(func(m *backuppb.BackupMeta) {
        m.StartVersion = req.StartVersion
        m.EndVersion = req.EndVersion
        m.IsRawKv = false
        m.IsTxnKv = true
        m.ClusterId = req.ClusterId
        m.ClusterVersion = mgr.GetClusterVersion(ctx)
        m.BrVersion = brVersion
        m.ApiVersion = client.GetApiVersion()
    })
    metaWriter.FinishWriteMetas(ctx, metautil.AppendDataFile)
    metaWriter.FlushBackupMeta(ctx)
}

BackupRanges // make a backup of the given key ranges.
  |-mainBackupLoop := &MainBackupLoop
  |     BackupSender:       &MainBackupSender{},
  |     BackupReq:          request,
  |     Concurrency:        concurrency,
  |     GlobalProgressTree: &globalProgressTree,
  |     ReplicaReadLabel:   replicaReadLabel,
  |     GetBackupClientCallBack: func(ctx , storeID uint64, reset bool) (backuppb.BackupClient, error) {
  |         return bc.mgr.GetBackupClient(ctx, storeID)
  |     },
  | }
  |-bc.RunLoop(ctx, mainBackupLoop) // infinite loop to backup ranges on all tikv stores
      |-for {
          inCompleteRanges = iter.GetIncompleteRanges() // 还未完成备份的 key 范围
          loop.BackupReq.SubRanges = getBackupRanges(inCompleteRanges)
          allStores := bc.getBackupStores(mainCtx, loop.ReplicaReadLabel)
          for _, store := range allStores {
            cli := loop.GetBackupClientCallBack(mainCtx, storeID, reset)
            loop.SendAsync(round, storeID, loop.BackupReq, loop.Concurrency, cli, ch, loop.StateNotifier)
                  |-go startBackup(storeID, request, cli, concurrency, respCh)
                        |-for i, req := range reqs {
                            doSendBackup(ectx, backupCli, bkReq, ...)
                              |-ctx, timerecv := StartTimeoutRecv(pctx, TimeoutOneResponse)
                              |-bCli := client.Backup(ctx, &req) // protobuf grpc method
                              |-for {
                              |-    resp := bCli.Recv()
                              |-    timerecv.Refresh()
                              |-    respFn(resp)
                              |-}
          }
        }

// components/backup/src/service.rs

impl<H> Backup for Service<H>
{
    fn backup(
        req: BackupRequest,
        mut sink: ServerStreamingSink<BackupResponse>,
    ) {
        if let Err(status) = match Task::new(req, tx) {
            Ok((task, c)) => {
                self.scheduler.schedule(task)
            }
        }

        let send_task = async move {
            let mut s = rx.map(|resp| Ok((resp, WriteFlags::default())));
            sink.send_all(&mut s).await?;
        }

        ctx.spawn(send_task);
    }
}

/// Backup Task.
pub struct Task {
    request: Request,
    pub(crate) resp: UnboundedSender<BackupResponse>,
}

// components/backup/src/endpoint.rs
impl Task {
    /// Create a backup task based on the given backup request.
    pub fn new(
        req: BackupRequest,
        resp: UnboundedSender<BackupResponse>,
    ) -> Result<(Task, Arc<AtomicBool>)> {
        let speed_limit = req.get_rate_limit();
        let limiter = Limiter::new(if speed_limit > 0  else f64::INFINITY });
        let cf = name_to_cf(req.get_cf())

        let task = Task {
            request: Request {
                start_key: req.get_start_key().to_owned(),
                end_key: req.get_end_key().to_owned(),
                sub_ranges: req.get_sub_ranges().to_owned(),
                start_ts: req.get_start_version().into(),
                end_ts: req.get_end_version().into(),
                backend: req.get_storage_backend().clone(),
                limiter,
                is_raw_kv: req.get_is_raw_kv(),
                dst_api_ver: req.get_dst_api_version(),
                cf,
                replica_read: req.get_replica_read(),
                resource_group_name: .get_resource_group_name().to_owned(),
                }),
            },
            resp,
        };
    }
}

// components/backup/src/endpoint.rs
BackupRanges -> BackupWriterBuilder -> S3Uploader

self.writer.put(&data_key_write, value) -> s3 put