implement peer snapshot client and associated methods

.github/workflows/golang.yml

@@ -38,22 +38,22 @@ jobs:
       fail-fast: false
       matrix:
         include:
-          - FABRIC_VERSION: "2.5.3"
+          - FABRIC_VERSION: "2.5.11"
             CREATE_CHANNEL: "create_channel"
             CONSENSUS: RAFT
-          - FABRIC_VERSION: "2.5.3"
+          - FABRIC_VERSION: "2.5.11"
             CREATE_CHANNEL: "existing_channel"
             CONSENSUS: RAFT
-          - FABRIC_VERSION: "3.0.0-preview"
+          - FABRIC_VERSION: "3.0.0"
             CREATE_CHANNEL: "create_channel"
             CONSENSUS: RAFT
-          - FABRIC_VERSION: "3.0.0-preview"
+          - FABRIC_VERSION: "3.0.0"
             CREATE_CHANNEL: "existing_channel"
             CONSENSUS: RAFT
-          - FABRIC_VERSION: "3.0.0-preview"
+          - FABRIC_VERSION: "3.0.0"
             CREATE_CHANNEL: "create_channel"
             CONSENSUS: BFT
-          - FABRIC_VERSION: "3.0.0-preview"
+          - FABRIC_VERSION: "3.0.0"
             CREATE_CHANNEL: "existing_channel"
             CONSENSUS: BFT
     steps:

pkg/snapshot/snapshot.go

@@ -0,0 +1,102 @@
+package snapshot
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/hyperledger/fabric-admin-sdk/internal/protoutil"
+	"github.com/hyperledger/fabric-admin-sdk/pkg/identity"
+	"github.com/hyperledger/fabric-protos-go-apiv2/common"
+	"github.com/hyperledger/fabric-protos-go-apiv2/peer"
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/proto"
+)
+
+// Peer is a wrapper around snapshot client
+type Peer struct {
+	snapshot peer.SnapshotClient
+	id       identity.SigningIdentity
+}
+
+func NewPeer(conn grpc.ClientConnInterface, id identity.SigningIdentity) *Peer {
+	return &Peer{
+		snapshot: peer.NewSnapshotClient(conn),
+		id:       id,
+	}
+}
+
+// SubmitRequest snapshot from a specific peer
+func (p *Peer) SubmitRequest(ctx context.Context, channelID string, blockNum uint64) error {
+	signedRequest, err := p.newSignedSnapshotRequest(channelID, blockNum)
+	if err != nil {
+		return fmt.Errorf("failed to create signed snapshot request: %w", err)
+	}
+
+	if _, err := p.snapshot.Generate(ctx, signedRequest); err != nil {
+		return fmt.Errorf("failed to submit generate snapshot request: %w", err)
+	}
+
+	return nil
+}
+
+// CancelRequest snapshot from a specific peer
+//
+// A snapshot for given blockNum must exist or an error will be returned
+func (p *Peer) CancelRequest(ctx context.Context, channelID string, blockNum uint64) error {
+	signedRequest, err := p.newSignedSnapshotRequest(channelID, blockNum)
+	if err != nil {
+		return fmt.Errorf("failed to create signed snapshot request: %w", err)
+	}
+
+	if _, err := p.snapshot.Cancel(ctx, signedRequest); err != nil {
+		return fmt.Errorf("failed to cancel snapshot request: %w", err)
+	}
+
+	return nil
+}
+
+// QueryPending snapshots from a specific peer
+func (p *Peer) QueryPending(ctx context.Context, channelID string) (*peer.QueryPendingSnapshotsResponse, error) {
+	signedRequest, err := p.newSignedSnapshotRequest(channelID, 0)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create signed snapshot request: %w", err)
+	}
+
+	res, err := p.snapshot.QueryPendings(ctx, signedRequest)
+	if err != nil {
+		return nil, fmt.Errorf("failed to list pending snapshots: %w", err)
+	}
+
+	return res, nil
+}
+
+// newSignedSnapshotRequest returns a signed snapshot request for
+// given channel
+func (p *Peer) newSignedSnapshotRequest(channelID string, blockNum uint64) (*peer.SignedSnapshotRequest, error) {
+	nonce, err := protoutil.CreateNonce()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create nonce: %w", err)
+	}
+
+	request, err := proto.Marshal(&peer.SnapshotRequest{
+		SignatureHeader: &common.SignatureHeader{
+			Creator: p.id.Credentials(),
+			Nonce:   nonce,
+		},
+		ChannelId:   channelID,
+		BlockNumber: blockNum,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal snapshot request: %w", err)
+	}
+
+	signature, err := p.id.Sign(request)
+	if err != nil {
+		return nil, fmt.Errorf("failed to sign snapshot request: %w", err)
+	}
+
+	return &peer.SignedSnapshotRequest{
+		Request:   request,
+		Signature: signature,
+	}, nil
+}

test/e2e_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/hyperledger/fabric-admin-sdk/pkg/discovery"
 	"github.com/hyperledger/fabric-admin-sdk/pkg/identity"
 	"github.com/hyperledger/fabric-admin-sdk/pkg/network"
+	"github.com/hyperledger/fabric-admin-sdk/pkg/snapshot"
 	"github.com/hyperledger/fabric-gateway/pkg/client"
 	gatewaypb "github.com/hyperledger/fabric-protos-go-apiv2/gateway"
 
@@ -34,6 +35,8 @@ const (
 	channelName     = "mychannel"
 	org1MspID       = "Org1MSP"
 	org2MspID       = "Org2MSP"
+
+	snapshotBlockNumber uint64 = 4
 )
 
 func runParallel[T any](args []T, f func(T)) {
@@ -389,6 +392,34 @@ var _ = Describe("e2e", func() {
 			Expect(err).NotTo(HaveOccurred())
 			Expect(peerMembershipResult.GetPeersByOrg()[org1MspID].GetPeers()).To(HaveLen(1))
 			Expect(peerMembershipResult.GetPeersByOrg()[org2MspID].GetPeers()).To(HaveLen(1))
+
+			// check snapshot
+			snapshotPeer := snapshot.NewPeer(peer1Connection, org1MSP)
+			snapshotCtx, snapshotCancel := context.WithTimeout(specCtx, 30*time.Second)
+			defer snapshotCancel()
+
+			// query snapshot when no snapshot request has been submitted
+			queryPendingRes, err := snapshotPeer.QueryPending(snapshotCtx, channelName)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(queryPendingRes.GetBlockNumbers()).To(BeEmpty(), "no pending snapshot requests before submission")
+
+			// submit snapshot request
+			err = snapshotPeer.SubmitRequest(snapshotCtx, channelName, snapshotBlockNumber)
+			Expect(err).NotTo(HaveOccurred(), "submit snapshot request")
+
+			// query pending snapshot request after submission
+			queryPendingRes, err = snapshotPeer.QueryPending(snapshotCtx, channelName)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(queryPendingRes.GetBlockNumbers()).To(ContainElement(snapshotBlockNumber), "pending snapshot request after submission")
+
+			// cancel snapshot request
+			err = snapshotPeer.CancelRequest(snapshotCtx, channelName, snapshotBlockNumber)
+			Expect(err).NotTo(HaveOccurred(), "cancel snapshot request after submission")
+
+			// query pending snapshot request after cancellation
+			queryPendingRes, err = snapshotPeer.QueryPending(snapshotCtx, channelName)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(queryPendingRes.GetBlockNumbers()).To(BeEmpty(), "no pending snapshot requests after cancellation")
 		})
 	})
 })