Merge branch 'main' into hm/snapshot-export

Author: hanabi1224

certstore/snapshot.go

@@ -15,6 +15,8 @@ import (
 var ErrUnknownLatestCertificate = errors.New("latest certificate is not known")
 
 // ExportLatestSnapshot exports an F3 snapshot that includes the finality certificate chain until the current `latestCertificate`.
+//
+// Checkout the format specification at <https://github.com/filecoin-project/FIPs/blob/master/FRCs/frc-0108.md>
 func (cs *Store) ExportLatestSnapshot(ctx context.Context, writer io.Writer) error {
 	if cs.latestCertificate == nil {
 		return ErrUnknownLatestCertificate
@@ -23,6 +25,8 @@ func (cs *Store) ExportLatestSnapshot(ctx context.Context, writer io.Writer) err
 }
 
 // ExportSnapshot exports an F3 snapshot that includes the finality certificate chain from the `Store.firstInstance` to the specified `lastInstance`.
+//
+// Checkout the format specification at <https://github.com/filecoin-project/FIPs/blob/master/FRCs/frc-0108.md>
 func (cs *Store) ExportSnapshot(ctx context.Context, latestInstance uint64, writer io.Writer) error {
 	initialPowerTable, err := cs.GetPowerTable(ctx, cs.firstInstance)
 	if err != nil {

certstore/snapshot_test.go

@@ -1 +1 @@
-package certstore
+package certstore

observer/model.go

@@ -1,7 +1,6 @@
 package observer
 
 import (
-	"encoding/base64"
 	"fmt"
 	"time"
 
@@ -151,11 +150,11 @@ func supplementalDataFromGpbft(sd gpbft.SupplementalData) SupplementalData {
 func (m Message) ToPartialMessage() (*gpbft.PartialGMessage, error) {
 	payload, err := m.Vote.ToGpbftPayload()
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("could not convert vote to gpbft payload: %w", err)
 	}
 	gj, err := m.Justification.ToGpbftJustification()
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("could not convert justification to gpbft justification: %w", err)
 	}
 
 	return &gpbft.PartialGMessage{
@@ -173,16 +172,16 @@ func (m Message) ToPartialMessage() (*gpbft.PartialGMessage, error) {
 func (p Payload) ToGpbftPayload() (gpbft.Payload, error) {
 	phase, err := phaseFromString(p.Phase)
 	if err != nil {
-		return gpbft.Payload{}, err
+		return gpbft.Payload{}, fmt.Errorf("could not convert phase: %w", err)
 	}
 	sd, err := p.SupplementalData.ToGpbftSupplementalData()
 	if err != nil {
-		return gpbft.Payload{}, err
+		return gpbft.Payload{}, fmt.Errorf("could not convert supplemental data: %w", err)
 	}
 
 	gv, err := ToGpbftChain(p.Value...)
 	if err != nil {
-		return gpbft.Payload{}, err
+		return gpbft.Payload{}, fmt.Errorf("could not convert value: %w", err)
 	}
 
 	return gpbft.Payload{
@@ -246,20 +245,13 @@ func (j *Justification) ToGpbftJustification() (*gpbft.Justification, error) {
 	gsigners, _ := bitfield.NewFromIter(ri)
 	payload, err := j.Vote.ToGpbftPayload()
 	if err != nil {
-		return nil, err
-	}
-
-	// Embedded struct as json, hence the decode.
-	s := string(j.Signature)
-	decoded, err := base64.StdEncoding.DecodeString(s)
-	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("could not convert vote to gpbft payload: %w", err)
 	}
 
 	return &gpbft.Justification{
 		Vote:      payload,
 		Signers:   gsigners,
-		Signature: decoded,
+		Signature: j.Signature,
 	}, nil
 }
 

observer/observer.go

@@ -215,7 +215,7 @@ func (o *Observer) createOrReplaceMessagesView(ctx context.Context, includeParqu
 	if includeParquet {
 		includeParquetFiles = fmt.Sprintf("UNION ALL SELECT * FROM '%s'", filepath.Join(o.rotatePath, "*.parquet"))
 	}
-	createView := fmt.Sprintf(`CREATE VIEW IF NOT EXISTS messages AS SELECT * FROM latest_messages %s`, includeParquetFiles)
+	createView := fmt.Sprintf(`CREATE OR REPLACE VIEW messages AS SELECT * FROM latest_messages %s`, includeParquetFiles)
 	_, err := o.db.ExecContext(ctx, createView)
 	return err