[Merged by Bors] - Support poet certificates

Makefile-libs.Inc

@@ -50,7 +50,7 @@ else
 	endif
 endif
 
-POSTRS_SETUP_REV = 0.7.7
+POSTRS_SETUP_REV = 0.7.8
 POSTRS_SETUP_ZIP = libpost-$(platform)-v$(POSTRS_SETUP_REV).zip
 POSTRS_SETUP_URL_ZIP ?= https://github.com/spacemeshos/post-rs/releases/download/v$(POSTRS_SETUP_REV)/$(POSTRS_SETUP_ZIP)
 

activation/activation.go

@@ -31,6 +31,8 @@
 	"github.com/spacemeshos/go-spacemesh/sql/localsql/nipost"
 )
 
+var ErrNotFound = errors.New("not found")
+
 // PoetConfig is the configuration to interact with the poet server.
 type PoetConfig struct {
 	PhaseShift        time.Duration `mapstructure:"phase-shift"`
@@ -77,7 +79,6 @@
 	localDB           *localsql.Database
 	publisher         pubsub.Publisher
 	nipostBuilder     nipostBuilder
-	certifier         certifierService
 	validator         nipostValidator
 	layerClock        layerClock
 	syncer            syncer
@@ -158,12 +159,6 @@
 	}
 }
 
-func WithPoetCertifier(c certifierService) BuilderOption {
-	return func(b *Builder) {
-		b.certifier = c
-	}
-}
-
 // NewBuilder returns an atx builder that will start a routine that will attempt to create an atx upon each new layer.
 func NewBuilder(
 	conf Config,
@@ -186,7 +181,6 @@
 		localDB:           localDB,
 		publisher:         publisher,
 		nipostBuilder:     nipostBuilder,
-		certifier:         &disabledCertifier{},
 		layerClock:        layerClock,
 		syncer:            syncer,
 		log:               log,
@@ -377,7 +371,7 @@
 	}, postInfo.NumUnits)
 	if err != nil {
 		b.log.Error("initial POST is invalid", log.ZShortStringer("smesherID", nodeID), zap.Error(err))
 		if err := nipost.RemovePost(b.localDB, nodeID); err != nil {
 			b.log.Fatal("failed to remove initial post", log.ZShortStringer("smesherID", nodeID), zap.Error(err))
 		}
 		return fmt.Errorf("initial POST is invalid: %w", err)
@@ -406,7 +400,17 @@
 		case <-b.layerClock.AwaitLayer(currentLayer.Add(1)):
 		}
 	}
-	b.certifier.CertifyAll(ctx, sig.NodeID(), b.poets)
+	var eg errgroup.Group
+	for _, poet := range b.poets {
+		eg.Go(func() error {
+			_, err := poet.Certify(ctx, sig.NodeID())
+			if err != nil {
+				b.log.Warn("failed to certify poet", zap.Error(err), log.ZShortStringer("smesherID", sig.NodeID()))
+			}
+			return nil
+		})
+	}
+	eg.Wait()
 
 	for {
 		err := b.PublishActivationTx(ctx, sig)
@@ -548,7 +552,7 @@
 		}, post.NumUnits)
 		if err != nil {
 			logger.Error("initial POST is invalid", zap.Error(err))
 			if err := nipost.RemovePost(b.localDB, nodeID); err != nil {
 				logger.Fatal("failed to remove initial post", zap.Error(err))
 			}
 			return nil, fmt.Errorf("initial POST is invalid: %w", err)

activation/certifier.go

@@ -9,20 +9,20 @@
 	"io"
 	"net/http"
 	"net/url"
+	"sync"
 	"time"
 
 	"github.com/hashicorp/go-retryablehttp"
 	"github.com/spacemeshos/poet/shared"
 	"go.uber.org/zap"
-	"golang.org/x/sync/errgroup"
 
 	"github.com/spacemeshos/go-spacemesh/activation/wire"
 	"github.com/spacemeshos/go-spacemesh/codec"
 	"github.com/spacemeshos/go-spacemesh/common/types"
 	"github.com/spacemeshos/go-spacemesh/sql"
 	"github.com/spacemeshos/go-spacemesh/sql/atxs"
 	"github.com/spacemeshos/go-spacemesh/sql/localsql"
-	"github.com/spacemeshos/go-spacemesh/sql/localsql/certifier"
+	certifierdb "github.com/spacemeshos/go-spacemesh/sql/localsql/certifier"
 	"github.com/spacemeshos/go-spacemesh/sql/localsql/nipost"
 )
 
@@ -82,6 +82,9 @@
 	logger *zap.Logger
 	db     *localsql.Database
 	client certifierClient
+
+	certificationsLock sync.Mutex
+	certifications     map[string]chan struct{}
 }
 
 func NewCertifier(
@@ -90,139 +93,67 @@
 	client certifierClient,
 ) *Certifier {
 	c := &Certifier{
-		client: client,
-		logger: logger,
-		db:     db,
+		client:         client,
+		logger:         logger,
+		db:             db,
+		certifications: make(map[string]chan struct{}),
 	}
 
 	return c
 }
 
-func (c *Certifier) Certificate(id types.NodeID, poet string) *certifier.PoetCert {
-	cert, err := certifier.Certificate(c.db, id, poet)
+func (c *Certifier) Certificate(
+	ctx context.Context,
+	id types.NodeID,
+	certifier *url.URL,
+	pubkey []byte,
+) (*certifierdb.PoetCert, error) {
+	// We index certs in DB by node ID and pubkey. To avoid redundant queries, we allow only 1
+	// request per (nodeID, pubkey) pair to be in flight at a time.
+	key := string(append(id.Bytes(), pubkey...))
+	c.certificationsLock.Lock()
+	if ch, ok := c.certifications[key]; ok {
+		c.certificationsLock.Unlock()
+		<-ch
+	} else {
+		ch := make(chan struct{})
+		c.certifications[key] = ch
+		c.certificationsLock.Unlock()
+		defer func() {
+			c.certificationsLock.Lock()
+			close(ch)
+			delete(c.certifications, key)
+			c.certificationsLock.Unlock()
+		}()
+	}
+
+	cert, err := certifierdb.Certificate(c.db, id, pubkey)
 	switch {
 	case err == nil:
-		return cert
+		return cert, nil
 	case !errors.Is(err, sql.ErrNotFound):
-		c.logger.Warn("failed to get certificate", zap.Error(err))
+		return nil, fmt.Errorf("getting certificate from DB for: %w", err)
 	}
-	return nil
+	return c.Recertify(ctx, id, certifier, pubkey)
 }
 
-func (c *Certifier) Recertify(ctx context.Context, id types.NodeID, poet PoetClient) (*certifier.PoetCert, error) {
-	url, pubkey, err := poet.CertifierInfo(ctx)
-	if err != nil {
-		return nil, fmt.Errorf("querying certifier info: %w", err)
-	}
-	cert, err := c.client.Certify(ctx, id, url, pubkey)
+func (c *Certifier) Recertify(
+	ctx context.Context,
+	id types.NodeID,
+	certifier *url.URL,
+	pubkey []byte,
+) (*certifierdb.PoetCert, error) {
+	cert, err := c.client.Certificate(ctx, id, certifier, pubkey)
 	if err != nil {
-		return nil, fmt.Errorf("certifying POST for %s at %v: %w", poet.Address(), url, err)
+		return nil, fmt.Errorf("certifying POST at %v: %w", certifier, err)
 	}
 
-	if err := certifier.AddCertificate(c.db, id, *cert, poet.Address()); err != nil {
+	if err := certifierdb.AddCertificate(c.db, id, *cert, pubkey); err != nil {
 		c.logger.Warn("failed to persist poet cert", zap.Error(err))
 	}
-
 	return cert, nil
 }
 
-// CertifyAll certifies the nodeID for all poets that require a certificate.
-// It optimizes the number of certification requests by taking a unique set of
-// certifiers among the given poets and sending a single request to each of them.
-// It returns a map of a poet address to a certificate for it.
-func (c *Certifier) CertifyAll(
-	ctx context.Context,
-	id types.NodeID,
-	poets []PoetClient,
-) map[string]*certifier.PoetCert {
-	certs := make(map[string]*certifier.PoetCert)
-	poetsToCertify := []PoetClient{}
-	for _, poet := range poets {
-		if cert := c.Certificate(id, poet.Address()); cert != nil {
-			certs[poet.Address()] = cert
-		} else {
-			poetsToCertify = append(poetsToCertify, poet)
-		}
-	}
-	if len(poetsToCertify) == 0 {
-		return certs
-	}
-
-	type certInfo struct {
-		url    *url.URL
-		pubkey []byte
-		poet   string
-	}
-
-	certifierInfos := make([]*certInfo, len(poetsToCertify))
-	var eg errgroup.Group
-	for i, poet := range poetsToCertify {
-		eg.Go(func() error {
-			url, pubkey, err := poet.CertifierInfo(ctx)
-			if err != nil {
-				c.logger.Warn("failed to query for certifier info", zap.Error(err), zap.String("poet", poet.Address()))
-				return nil
-			}
-			certifierInfos[i] = &certInfo{
-				url:    url,
-				pubkey: pubkey,
-				poet:   poet.Address(),
-			}
-			return nil
-		})
-	}
-	eg.Wait()
-
-	type certService struct {
-		url    *url.URL
-		pubkey []byte
-		poets  []string
-	}
-	certSvcs := make(map[string]*certService)
-	for _, info := range certifierInfos {
-		if info == nil {
-			continue
-		}
-
-		if svc, ok := certSvcs[string(info.pubkey)]; !ok {
-			certSvcs[string(info.pubkey)] = &certService{
-				url:    info.url,
-				pubkey: info.pubkey,
-				poets:  []string{info.poet},
-			}
-		} else {
-			svc.poets = append(svc.poets, info.poet)
-		}
-	}
-
-	for _, svc := range certSvcs {
-		c.logger.Info(
-			"certifying for poets",
-			zap.Stringer("certifier", svc.url),
-			zap.Strings("poets", svc.poets),
-		)
-
-		cert, err := c.client.Certify(ctx, id, svc.url, svc.pubkey)
-		if err != nil {
-			c.logger.Warn("failed to certify", zap.Error(err), zap.Stringer("certifier", svc.url))
-			continue
-		}
-		c.logger.Info(
-			"successfully obtained certificate",
-			zap.Stringer("certifier", svc.url),
-			zap.Binary("cert data", cert.Data),
-			zap.Binary("cert signature", cert.Signature),
-		)
-		for _, poet := range svc.poets {
-			if err := certifier.AddCertificate(c.db, id, *cert, poet); err != nil {
-				c.logger.Warn("failed to persist poet cert", zap.Error(err))
-			}
-			certs[poet] = cert
-		}
-	}
-	return certs
-}
-
 type CertifierClient struct {
 	client  *retryablehttp.Client
 	logger  *zap.Logger
@@ -275,17 +206,17 @@
 	}
 	atx, err := atxs.Get(c.db, atxid)
 	if err != nil {
 		return nil, fmt.Errorf("failed to retrieve ATX: %w", err)
 	}
 	atxNipost, err := loadNipost(ctx, c.db, atxid)
 	if err != nil {
 		return nil, errors.New("no NIPoST found in last ATX")
 	}
 	if atx.CommitmentATX == nil {
 		if commitmentAtx, err := atxs.CommitmentATX(c.db, nodeId); err != nil {
 			return nil, fmt.Errorf("failed to retrieve commitment ATX: %w", err)
 		} else {
 			atx.CommitmentATX = &commitmentAtx
 		}
 	}
 
@@ -310,15 +241,15 @@
 		return post, nil
 	case errors.Is(err, sql.ErrNotFound):
 		// no post found
 	default:
 		return nil, fmt.Errorf("loading initial post from db: %w", err)
 	}
 
 	c.logger.Info("POST not found in local DB. Trying to obtain POST from an existing ATX")
 	if post, err := c.obtainPostFromLastAtx(ctx, id); err == nil {
 		c.logger.Info("found POST in an existing ATX")
 		if err := nipost.AddPost(c.localDb, id, *post); err != nil {
 			c.logger.Error("failed to save post", zap.Error(err))
 		}
 		return post, nil
 	}
@@ -326,15 +257,15 @@
 	return nil, errors.New("PoST not found")
 }
 
-func (c *CertifierClient) Certify(
+func (c *CertifierClient) Certificate(
 	ctx context.Context,
 	id types.NodeID,
 	url *url.URL,
 	pubkey []byte,
-) (*certifier.PoetCert, error) {
+) (*certifierdb.PoetCert, error) {
 	post, err := c.obtainPost(ctx, id)
 	if err != nil {
 		return nil, fmt.Errorf("obtaining PoST: %w", err)
 	}
 
 	request := CertifyRequest{
@@ -353,35 +284,35 @@
 
 	jsonRequest, err := json.Marshal(request)
 	if err != nil {
 		return nil, fmt.Errorf("marshaling request: %w", err)
 	}
 
 	req, err := retryablehttp.NewRequestWithContext(ctx, "POST", url.JoinPath("/certify").String(), jsonRequest)
 	if err != nil {
 		return nil, fmt.Errorf("creating HTTP request: %w", err)
 	}
 	req.Header.Set("Content-Type", "application/json")
 
 	resp, err := c.client.Do(req)
 	if err != nil {
 		return nil, fmt.Errorf("sending request: %w", err)
 	}
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
 		var body []byte
 		if resp.Body != nil {
 			body, _ = io.ReadAll(resp.Body)
 		}
 		return nil, fmt.Errorf("request failed with code %d (message: %s)", resp.StatusCode, body)
 	}
 
 	certResponse := CertifyResponse{}
 	if err := json.NewDecoder(resp.Body).Decode(&certResponse); err != nil {
 		return nil, fmt.Errorf("decoding JSON response: %w", err)
 	}
 	if !bytes.Equal(certResponse.PubKey, pubkey) {
 		return nil, errors.New("pubkey is invalid")
 	}
 
 	opaqueCert := &shared.OpaqueCert{
@@ -397,41 +328,33 @@
 	if cert.Expiration != nil {
 		c.logger.Info("certificate has expiration date", zap.Time("expiration", *cert.Expiration))
 		if time.Until(*cert.Expiration) < 0 {
 			return nil, errors.New("certificate is expired")
 		}
 	}
 
-	return &certifier.PoetCert{
+	return &certifierdb.PoetCert{
 		Data:      opaqueCert.Data,
 		Signature: opaqueCert.Signature,
 	}, nil
 }
 
-type disabledCertifier struct{}
-
-func (d *disabledCertifier) Certificate(types.NodeID, string) *certifier.PoetCert {
-	return nil
-}
-
-func (d *disabledCertifier) Recertify(context.Context, types.NodeID, PoetClient) (*certifier.PoetCert, error) {
-	return nil, errors.New("certifier disabled")
-}
-
-func (d *disabledCertifier) CertifyAll(context.Context, types.NodeID, []PoetClient) map[string]*certifier.PoetCert {
-	return nil
-}
-
 // load NIPoST for the given ATX from the database.
 func loadNipost(ctx context.Context, db sql.Executor, id types.ATXID) (*types.NIPost, error) {
 	var blob sql.Blob
-	if err := atxs.LoadBlob(ctx, db, id.Bytes(), &blob); err != nil {
+	version, err := atxs.LoadBlob(ctx, db, id.Bytes(), &blob)
+	if err != nil {
 		return nil, fmt.Errorf("getting blob for %s: %w", id, err)
 	}
 
-	// TODO: decide about version based on the `version` column
-	var atx wire.ActivationTxV1
-	if err := codec.Decode(blob.Bytes, &atx); err != nil {
-		return nil, fmt.Errorf("decoding ATX blob: %w", err)
+	switch version {
+	case types.AtxV1:
+		var atx wire.ActivationTxV1
+		if err := codec.Decode(blob.Bytes, &atx); err != nil {
+			return nil, fmt.Errorf("decoding ATX blob: %w", err)
+		}
+		return wire.NiPostFromWireV1(atx.NIPost), nil
+	case types.AtxV2:
+		// TODO: support ATX V2
 	}
-	return wire.NiPostFromWireV1(atx.NIPost), nil
+	panic("unsupported ATX version")
 }