Update to protobuf v1.25 and grpc 1.35; bump CI to go1.12

From:
github.com/golang/protobuf v1.3.2
google.golang.org/grpc v1.17.0

To:
github.com/golang/protobuf v1.4.3
google.golang.org/grpc v1.35.0
google.golang.org/protobuf v1.25.0

About the two protobuf packages:
https://developers.google.com/protocol-buffers/docs/reference/go/faq
> Version v1.4.0 and higher of github.com/golang/protobuf wrap the new
implementation and permit programs to adopt the new API incrementally. For
example, the well-known types defined in github.com/golang/protobuf/ptypes are
simply aliases of those defined in the newer module. Thus,
google.golang.org/protobuf/types/known/emptypb and
github.com/golang/protobuf/ptypes/empty may be used interchangeably.

Notable Code Changes in zrepl:
- generate protobufs now contain a mutex so we can't copy them by value
  anymore
- grpc.WithDialer is deprecated => use grpc.WithContextDialer instead

Go1.12 is now actually required by some of the dependencies.

replication/logic/pdu/pdu.proto

@@ -1,5 +1,5 @@
 syntax = "proto3";
-option go_package = "pdu";
+option go_package = ".;pdu";
 
 service Replication {
   rpc Ping(PingReq) returns (PingRes);

replication/logic/pdu/pdu_grpc.pb.go

@@ -0,0 +1,281 @@
+// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+
+package pdu
+
+import (
+	context "context"
+	grpc "google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	status "google.golang.org/grpc/status"
+)
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+// Requires gRPC-Go v1.32.0 or later.
+const _ = grpc.SupportPackageIsVersion7
+
+// ReplicationClient is the client API for Replication service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
+type ReplicationClient interface {
+	Ping(ctx context.Context, in *PingReq, opts ...grpc.CallOption) (*PingRes, error)
+	ListFilesystems(ctx context.Context, in *ListFilesystemReq, opts ...grpc.CallOption) (*ListFilesystemRes, error)
+	ListFilesystemVersions(ctx context.Context, in *ListFilesystemVersionsReq, opts ...grpc.CallOption) (*ListFilesystemVersionsRes, error)
+	DestroySnapshots(ctx context.Context, in *DestroySnapshotsReq, opts ...grpc.CallOption) (*DestroySnapshotsRes, error)
+	ReplicationCursor(ctx context.Context, in *ReplicationCursorReq, opts ...grpc.CallOption) (*ReplicationCursorRes, error)
+	SendCompleted(ctx context.Context, in *SendCompletedReq, opts ...grpc.CallOption) (*SendCompletedRes, error)
+}
+
+type replicationClient struct {
+	cc grpc.ClientConnInterface
+}
+
+func NewReplicationClient(cc grpc.ClientConnInterface) ReplicationClient {
+	return &replicationClient{cc}
+}
+
+func (c *replicationClient) Ping(ctx context.Context, in *PingReq, opts ...grpc.CallOption) (*PingRes, error) {
+	out := new(PingRes)
+	err := c.cc.Invoke(ctx, "/Replication/Ping", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *replicationClient) ListFilesystems(ctx context.Context, in *ListFilesystemReq, opts ...grpc.CallOption) (*ListFilesystemRes, error) {
+	out := new(ListFilesystemRes)
+	err := c.cc.Invoke(ctx, "/Replication/ListFilesystems", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *replicationClient) ListFilesystemVersions(ctx context.Context, in *ListFilesystemVersionsReq, opts ...grpc.CallOption) (*ListFilesystemVersionsRes, error) {
+	out := new(ListFilesystemVersionsRes)
+	err := c.cc.Invoke(ctx, "/Replication/ListFilesystemVersions", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *replicationClient) DestroySnapshots(ctx context.Context, in *DestroySnapshotsReq, opts ...grpc.CallOption) (*DestroySnapshotsRes, error) {
+	out := new(DestroySnapshotsRes)
+	err := c.cc.Invoke(ctx, "/Replication/DestroySnapshots", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *replicationClient) ReplicationCursor(ctx context.Context, in *ReplicationCursorReq, opts ...grpc.CallOption) (*ReplicationCursorRes, error) {
+	out := new(ReplicationCursorRes)
+	err := c.cc.Invoke(ctx, "/Replication/ReplicationCursor", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *replicationClient) SendCompleted(ctx context.Context, in *SendCompletedReq, opts ...grpc.CallOption) (*SendCompletedRes, error) {
+	out := new(SendCompletedRes)
+	err := c.cc.Invoke(ctx, "/Replication/SendCompleted", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// ReplicationServer is the server API for Replication service.
+// All implementations must embed UnimplementedReplicationServer
+// for forward compatibility
+type ReplicationServer interface {
+	Ping(context.Context, *PingReq) (*PingRes, error)
+	ListFilesystems(context.Context, *ListFilesystemReq) (*ListFilesystemRes, error)
+	ListFilesystemVersions(context.Context, *ListFilesystemVersionsReq) (*ListFilesystemVersionsRes, error)
+	DestroySnapshots(context.Context, *DestroySnapshotsReq) (*DestroySnapshotsRes, error)
+	ReplicationCursor(context.Context, *ReplicationCursorReq) (*ReplicationCursorRes, error)
+	SendCompleted(context.Context, *SendCompletedReq) (*SendCompletedRes, error)
+	mustEmbedUnimplementedReplicationServer()
+}
+
+// UnimplementedReplicationServer must be embedded to have forward compatible implementations.
+type UnimplementedReplicationServer struct {
+}
+
+func (UnimplementedReplicationServer) Ping(context.Context, *PingReq) (*PingRes, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method Ping not implemented")
+}
+func (UnimplementedReplicationServer) ListFilesystems(context.Context, *ListFilesystemReq) (*ListFilesystemRes, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method ListFilesystems not implemented")
+}
+func (UnimplementedReplicationServer) ListFilesystemVersions(context.Context, *ListFilesystemVersionsReq) (*ListFilesystemVersionsRes, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method ListFilesystemVersions not implemented")
+}
+func (UnimplementedReplicationServer) DestroySnapshots(context.Context, *DestroySnapshotsReq) (*DestroySnapshotsRes, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method DestroySnapshots not implemented")
+}
+func (UnimplementedReplicationServer) ReplicationCursor(context.Context, *ReplicationCursorReq) (*ReplicationCursorRes, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method ReplicationCursor not implemented")
+}
+func (UnimplementedReplicationServer) SendCompleted(context.Context, *SendCompletedReq) (*SendCompletedRes, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method SendCompleted not implemented")
+}
+func (UnimplementedReplicationServer) mustEmbedUnimplementedReplicationServer() {}
+
+// UnsafeReplicationServer may be embedded to opt out of forward compatibility for this service.
+// Use of this interface is not recommended, as added methods to ReplicationServer will
+// result in compilation errors.
+type UnsafeReplicationServer interface {
+	mustEmbedUnimplementedReplicationServer()
+}
+
+func RegisterReplicationServer(s grpc.ServiceRegistrar, srv ReplicationServer) {
+	s.RegisterService(&Replication_ServiceDesc, srv)
+}
+
+func _Replication_Ping_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(PingReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ReplicationServer).Ping(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/Replication/Ping",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ReplicationServer).Ping(ctx, req.(*PingReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Replication_ListFilesystems_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ListFilesystemReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ReplicationServer).ListFilesystems(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/Replication/ListFilesystems",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ReplicationServer).ListFilesystems(ctx, req.(*ListFilesystemReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Replication_ListFilesystemVersions_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ListFilesystemVersionsReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ReplicationServer).ListFilesystemVersions(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/Replication/ListFilesystemVersions",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ReplicationServer).ListFilesystemVersions(ctx, req.(*ListFilesystemVersionsReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Replication_DestroySnapshots_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(DestroySnapshotsReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ReplicationServer).DestroySnapshots(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/Replication/DestroySnapshots",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ReplicationServer).DestroySnapshots(ctx, req.(*DestroySnapshotsReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Replication_ReplicationCursor_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ReplicationCursorReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ReplicationServer).ReplicationCursor(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/Replication/ReplicationCursor",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ReplicationServer).ReplicationCursor(ctx, req.(*ReplicationCursorReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Replication_SendCompleted_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(SendCompletedReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ReplicationServer).SendCompleted(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/Replication/SendCompleted",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ReplicationServer).SendCompleted(ctx, req.(*SendCompletedReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+// Replication_ServiceDesc is the grpc.ServiceDesc for Replication service.
+// It's only intended for direct use with grpc.RegisterService,
+// and not to be introspected or modified (even as a copy)
+var Replication_ServiceDesc = grpc.ServiceDesc{
+	ServiceName: "Replication",
+	HandlerType: (*ReplicationServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "Ping",
+			Handler:    _Replication_Ping_Handler,
+		},
+		{
+			MethodName: "ListFilesystems",
+			Handler:    _Replication_ListFilesystems_Handler,
+		},
+		{
+			MethodName: "ListFilesystemVersions",
+			Handler:    _Replication_ListFilesystemVersions_Handler,
+		},
+		{
+			MethodName: "DestroySnapshots",
+			Handler:    _Replication_DestroySnapshots_Handler,
+		},
+		{
+			MethodName: "ReplicationCursor",
+			Handler:    _Replication_ReplicationCursor_Handler,
+		},
+		{
+			MethodName: "SendCompleted",
+			Handler:    _Replication_SendCompleted_Handler,
+		},
+	},
+	Streams:  []grpc.StreamDesc{},
+	Metadata: "pdu.proto",
+}

replication/logic/pdu/pdu_test.go

@@ -10,7 +10,7 @@ import (
 func TestFilesystemVersion_RelName(t *testing.T) {
 
 	type TestCase struct {
-		In    FilesystemVersion
+		In    *FilesystemVersion
 		Out   string
 		Panic bool
 	}
@@ -18,7 +18,7 @@ func TestFilesystemVersion_RelName(t *testing.T) {
 	creat := FilesystemVersionCreation(time.Now())
 	tcs := []TestCase{
 		{
-			In: FilesystemVersion{
+			In: &FilesystemVersion{
 				Type:     FilesystemVersion_Snapshot,
 				Name:     "foobar",
 				Creation: creat,
@@ -26,7 +26,7 @@ func TestFilesystemVersion_RelName(t *testing.T) {
 			Out: "@foobar",
 		},
 		{
-			In: FilesystemVersion{
+			In: &FilesystemVersion{
 				Type:     FilesystemVersion_Bookmark,
 				Name:     "foobar",
 				Creation: creat,
@@ -34,7 +34,7 @@ func TestFilesystemVersion_RelName(t *testing.T) {
 			Out: "#foobar",
 		},
 		{
-			In: FilesystemVersion{
+			In: &FilesystemVersion{
 				Type:     2342,
 				Name:     "foobar",
 				Creation: creat,

replication/logic/replication_logic.go

@@ -117,7 +117,7 @@ func (p *Planner) WaitForConnectivity(ctx context.Context) error {
 type Filesystem struct {
 	sender   Sender
 	receiver Receiver
-	policy   PlannerPolicy
+	policy   PlannerPolicy // immutable, it's .ReplicationConfig member is a pointer and copied into messages
 
 	Path                 string             // compat
 	receiverFS, senderFS *pdu.Filesystem    // receiverFS may be nil, senderFS never nil
@@ -569,7 +569,7 @@ func (s *Step) buildSendRequest(dryRun bool) (sr *pdu.SendReq) {
 		Encrypted:         s.encrypt.ToPDU(),
 		ResumeToken:       s.resumeToken,
 		DryRun:            dryRun,
-		ReplicationConfig: &s.parent.policy.ReplicationConfig,
+		ReplicationConfig: s.parent.policy.ReplicationConfig,
 	}
 	return sr
 }
@@ -614,7 +614,7 @@ func (s *Step) doReplication(ctx context.Context) error {
 		Filesystem:        fs,
 		To:                sr.GetTo(),
 		ClearResumeToken:  !sres.UsedResumeToken,
-		ReplicationConfig: &s.parent.policy.ReplicationConfig,
+		ReplicationConfig: s.parent.policy.ReplicationConfig,
 	}
 	log.Debug("initiate receive request")
 	_, err = s.receiver.Receive(ctx, rr, byteCountingStream)

replication/logic/replication_logic_policy.go

@@ -9,7 +9,7 @@ import (
 
 type PlannerPolicy struct {
 	EncryptedSend     tri // all sends must be encrypted (send -w, and encryption!=off)
-	ReplicationConfig pdu.ReplicationConfig
+	ReplicationConfig *pdu.ReplicationConfig
 }
 
 func ReplicationConfigFromConfig(in *config.Replication) (*pdu.ReplicationConfig, error) {