Обновить internal/endpoint/endpoint.go

internal/endpoint/endpoint.go

@@ -106,6 +106,7 @@ func (s *Sender) ListFilesystems(ctx context.Context, r *pdu.ListFilesystemReq)
 	}
 	rfss := make([]*pdu.Filesystem, 0, len(fss))
 	for _, a := range fss {
+		// TODO: dedup code with Receiver.ListFilesystems
 		l := getLogger(ctx).WithField("fs", a)
 		ph, err := zfs.ZFSGetFilesystemPlaceholderState(ctx, a)
 		if err != nil {
@@ -119,14 +120,9 @@ func (s *Sender) ListFilesystems(ctx context.Context, r *pdu.ListFilesystemReq)
 			return nil, err
 		}
 
-		// === ПАТЧ: убираем pool name перед отправкой receiver'у ===
-		p := a.Copy()
-		if p.Length() > 1 {
-			p.TrimNPrefixComps(1)
-		}
-
 		fs := &pdu.Filesystem{
-			Path:          p.ToString(),
+			Path: a.ToString(),
+			// ResumeToken does not make sense from Sender
 			IsPlaceholder: ph.IsPlaceholder,
 		}
 		rfss = append(rfss, fs)
@@ -135,26 +131,24 @@ func (s *Sender) ListFilesystems(ctx context.Context, r *pdu.ListFilesystemReq)
 	return res, nil
 }
 
-func (s *Receiver) ListFilesystemVersions(ctx context.Context, req *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error) {
+func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error) {
 	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 
-	root := s.clientRootFromCtx(ctx)
+	lp, err := s.filterCheckFS(r.GetFilesystem())
-	lp, err := subroot{root}.MapToLocal(req.GetFilesystem())
 	if err != nil {
 		return nil, err
 	}
-
 	fsvs, err := zfs.ZFSListFilesystemVersions(ctx, lp, zfs.ListFilesystemVersionsOptions{})
 	if err != nil {
 		return nil, err
 	}
-
 	rfsvs := make([]*pdu.FilesystemVersion, len(fsvs))
 	for i := range fsvs {
 		rfsvs[i] = pdu.FilesystemVersionFromZFS(&fsvs[i])
 	}
+	res := &pdu.ListFilesystemVersionsRes{Versions: rfsvs}
+	return res, nil
 
-	return &pdu.ListFilesystemVersionsRes{Versions: rfsvs}, nil
 }
 
 func uncheckedSendArgsFromPDU(fsv *pdu.FilesystemVersion) *zfs.ZFSSendArgVersion {
@@ -602,8 +596,26 @@ func clientRoot(rootFS *zfs.DatasetPath, clientIdentity string) (*zfs.DatasetPat
 }
 
 func (s *Receiver) clientRootFromCtx(ctx context.Context) *zfs.DatasetPath {
-	// Всегда чистый root без client identity
+	if !s.conf.AppendClientIdentity {
-	return s.conf.RootWithoutClientComponent.Copy()
+		return s.conf.RootWithoutClientComponent.Copy()
+	}
+
+	var clientIdentity string
+	if s.Test_OverrideClientIdentityFunc != nil {
+		clientIdentity = s.Test_OverrideClientIdentityFunc()
+	} else {
+		var ok bool
+		clientIdentity, ok = ctx.Value(ClientIdentityKey).(string) // no shadow
+		if !ok {
+			panic("ClientIdentityKey context value must be set")
+		}
+	}
+
+	clientRoot, err := clientRoot(s.conf.RootWithoutClientComponent, clientIdentity)
+	if err != nil {
+		panic(fmt.Sprintf("ClientIdentityContextKey must have been validated before invoking Receiver: %s", err))
+	}
+	return clientRoot
 }
 
 type subroot struct {
@@ -623,6 +635,7 @@ func (f subroot) UserSpecifiedDatasets() zfs.UserSpecifiedDatasetsSet {
 	}
 }
 
+
 func (f subroot) MapToLocal(fs string) (*zfs.DatasetPath, error) {
 	p, err := zfs.NewDatasetPath(fs)
 	if err != nil {
@@ -631,26 +644,18 @@ func (f subroot) MapToLocal(fs string) (*zfs.DatasetPath, error) {
 	if p.Length() == 0 {
 		return nil, errors.Errorf("cannot map empty filesystem")
 	}
-
+	// drop N leading components of the sender's path (e.g. pool name)
-	// Убираем pool из source
+	if n := envconst.Int("ZREPL_STRIP_SENDER_PREFIX_COMPS", 0); n > 0 {
-	if p.Length() > 1 {
+		p.TrimNPrefixComps(n)
-		subpath := p.Copy()
-		subpath.TrimNPrefixComps(1)
-		p = subpath
 	}
-
-	// Оставляем только chunk_9XG1J8T0
 	c := f.localRoot.Copy()
-	if c.Length() > 1 {
-		rootOnly := c.Copy()
-		rootOnly.TrimNPrefixComps(c.Length() - 1)
-		c = rootOnly
-	}
-
 	c.Extend(p)
 	return c, nil
 }
 
+
+
+
 func (s *Receiver) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) {
 	defer trace.WithSpanFromStackUpdateCtx(&ctx)()