Merge pull request #67 from qubic/compression

Compression

rpc/rpc_server.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/hex"
 	"encoding/json"
+	"fmt"
 	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
 	"github.com/pkg/errors"
 	"github.com/qubic/go-archiver/protobuff"
@@ -238,6 +239,27 @@ func (s *Server) GetQuorumTickData(ctx context.Context, req *protobuff.GetQuorum
 		return nil, status.Errorf(codes.Internal, "getting processed tick intervals per epoch")
 	}
 
+	epoch, err := getTickEpoch(req.TickNumber, processedTickIntervalsPerEpoch)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "getting tick epoch :%v", err)
+	}
+
+	lastTickFlag, err := isLastTick(req.TickNumber, epoch, processedTickIntervalsPerEpoch)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "checking if tick is last tick in it's epoch: %v", err)
+	}
+
+	if lastTickFlag {
+		lastTickQuorumData, err := s.store.GetLastTickQuorumDataPerEpoch(epoch)
+		if err != nil {
+			return nil, status.Errorf(codes.Internal, "getting quorum data for last processed tick: %v", err)
+		}
+
+		return &protobuff.GetQuorumTickDataResponse{
+			QuorumTickData: lastTickQuorumData,
+		}, nil
+	}
+
 	wasSkipped, nextAvailableTick := wasTickSkippedByArchive(req.TickNumber, processedTickIntervalsPerEpoch)
 	if wasSkipped == true {
 		st := status.Newf(codes.OutOfRange, "provided tick number %d was skipped by the system, next available tick is %d", req.TickNumber, nextAvailableTick)
@@ -537,6 +559,48 @@ func wasTickSkippedByArchive(tick uint32, processedTicksIntervalPerEpoch []*prot
 	return false, 0
 }
 
+func getTickEpoch(tickNumber uint32, intervals []*protobuff.ProcessedTickIntervalsPerEpoch) (uint32, error) {
+	if len(intervals) == 0 {
+		return 0, errors.New("processed tick interval list is empty")
+	}
+
+	for _, epochInterval := range intervals {
+		for _, interval := range epochInterval.Intervals {
+			if tickNumber >= interval.InitialProcessedTick && tickNumber <= interval.LastProcessedTick {
+				return epochInterval.Epoch, nil
+			}
+		}
+	}
+
+	return 0, errors.New(fmt.Sprintf("unable to find the epoch for tick %d", tickNumber))
+}
+
+func getProcessedTickIntervalsForEpoch(epoch uint32, intervals []*protobuff.ProcessedTickIntervalsPerEpoch) (*protobuff.ProcessedTickIntervalsPerEpoch, error) {
+	for _, interval := range intervals {
+		if interval.Epoch != epoch {
+			continue
+		}
+		return interval, nil
+	}
+
+	return nil, errors.New(fmt.Sprintf("unable to find processed tick intervals for epoch %d", epoch))
+}
+
+func isLastTick(tickNumber uint32, epoch uint32, intervals []*protobuff.ProcessedTickIntervalsPerEpoch) (bool, error) {
+	epochIntervals, err := getProcessedTickIntervalsForEpoch(epoch, intervals)
+	if err != nil {
+		return false, errors.Wrap(err, "getting processed tick intervals per epoch")
+	}
+
+	for _, interval := range epochIntervals.Intervals {
+		if interval.LastProcessedTick == tickNumber {
+			return true, nil
+		}
+	}
+
+	return false, nil
+}
+
 func (s *Server) GetTransactionStatus(ctx context.Context, req *protobuff.GetTransactionStatusRequest) (*protobuff.GetTransactionStatusResponse, error) {
 	id := types.Identity(req.TxId)
 	pubKey, err := id.ToPubKey(true)

store/keys.go

@@ -19,8 +19,15 @@ const (
 	TransactionStatus            = 0x11
 	StoreDigest                  = 0x12
 	EmptyTicksPerEpoch           = 0x13
+	LastTickQuorumDataPerEpoch   = 0x14
 )
 
+func lastTickQuorumDataPerEpochKey(epoch uint32) []byte {
+	key := []byte{LastTickQuorumDataPerEpoch}
+	key = binary.BigEndian.AppendUint64(key, uint64(epoch))
+	return key
+}
+
 func emptyTicksPerEpochKey(epoch uint32) []byte {
 	key := []byte{EmptyTicksPerEpoch}
 	key = binary.BigEndian.AppendUint64(key, uint64(epoch))

store/store.go

@@ -698,3 +698,40 @@ func (s *PebbleStore) DeleteEmptyTicksKeyForEpoch(epoch uint32) error {
 	}
 	return nil
 }
+
+func (s *PebbleStore) SetLastTickQuorumDataPerEpoch(quorumData *protobuff.QuorumTickData, epoch uint32) error {
+	key := lastTickQuorumDataPerEpochKey(epoch)
+
+	serialized, err := proto.Marshal(quorumData)
+	if err != nil {
+		return errors.Wrapf(err, "serializing quorum tick data for last tick of epoch %d", epoch)
+	}
+
+	err = s.db.Set(key, serialized, pebble.Sync)
+	if err != nil {
+		return errors.Wrapf(err, "setting last tick quorum tick data for epoch %d", epoch)
+	}
+	return nil
+}
+
+func (s *PebbleStore) GetLastTickQuorumDataPerEpoch(epoch uint32) (*protobuff.QuorumTickData, error) {
+	key := lastTickQuorumDataPerEpochKey(epoch)
+
+	value, closer, err := s.db.Get(key)
+	if err != nil {
+		if errors.Is(err, pebble.ErrNotFound) {
+			return nil, err
+		}
+		return nil, errors.Wrapf(err, "getting last tick quorum data for epoch %d", epoch)
+	}
+	defer closer.Close()
+
+	var quorumData *protobuff.QuorumTickData
+
+	err = proto.Unmarshal(value, quorumData)
+	if err != nil {
+		return nil, errors.Wrapf(err, "deserializing quorum tick data for last tick of epoch %d", epoch)
+	}
+
+	return quorumData, nil
+}

validator/quorum/validator.go

@@ -163,5 +163,12 @@ func Store(ctx context.Context, store *store.PebbleStore, tickNumber uint32, quo
 		return errors.Wrap(err, "set quorum votes")
 	}
 
+	fullProtoModel := qubicToProto(quorumVotes)
+
+	err = store.SetLastTickQuorumDataPerEpoch(fullProtoModel, protoModel.QuorumTickStructure.Epoch)
+	if err != nil {
+		return errors.Wrap(err, "setting last quorum tick data")
+	}
+
 	return nil
 }