[Merged by Bors] - Support poet certificates

activation/activation.go

@@ -75,8 +75,8 @@
 	localDB           *localsql.Database
 	publisher         pubsub.Publisher
 	nipostBuilder     nipostBuilder
+	certifier         certifierService
 	validator         nipostValidator
-	certifierConfig   CertifierConfig
 	layerClock        layerClock
 	syncer            syncer
 	log               *zap.Logger
@@ -94,7 +94,6 @@
 	// since they (can) modify the fields below.
 	smeshingMutex sync.Mutex
 	signers       map[types.NodeID]*signing.EdSigner
-	certifiers    map[types.NodeID]certifierService
 	eg            errgroup.Group
 	stop          context.CancelFunc
 }
@@ -147,9 +146,9 @@
 	}
 }
 
-func WithCertifierConfig(c CertifierConfig) BuilderOption {
+func WithPoetCertifier(c certifierService) BuilderOption {
 	return func(b *Builder) {
-		b.certifierConfig = c
+		b.certifier = c
 	}
 }
 
@@ -173,14 +172,13 @@
 		localDB:           localDB,
 		publisher:         publisher,
 		nipostBuilder:     nipostBuilder,
+		certifier:         &disabledCertifier{},
 		layerClock:        layerClock,
 		syncer:            syncer,
 		log:               log,
 		poetRetryInterval: defaultPoetRetryInterval,
 		postValidityDelay: 12 * time.Hour,
 		postStates:        NewPostStates(log),
-		certifiers:        make(map[types.NodeID]certifierService),
-		certifierConfig:   DefaultCertifierConfig(),
 	}
 	for _, opt := range opts {
 		opt(b)
@@ -320,18 +318,34 @@
 	return maps.Keys(b.signers)
 }
 
-// Create the initial post and save it.
-func (b *Builder) buildInitialPost(ctx context.Context, nodeID types.NodeID) (*types.Post, *types.PostInfo, error) {
+func (b *Builder) buildInitialPost(ctx context.Context, nodeID types.NodeID) error {
+	// Generate the initial POST if we don't have an ATX...
+	if _, err := atxs.GetLastIDByNodeID(b.db, nodeID); err == nil {
+		return nil
+	}
+	// ...and if we haven't stored an initial post yet.
+	_, err := nipost.GetPost(b.localDB, nodeID)
+	switch {
+	case err == nil:
+		b.log.Info("load initial post from db")
+		return nil
+	case errors.Is(err, sql.ErrNotFound):
+		b.log.Info("creating initial post")
+	default:
+		return fmt.Errorf("get initial post: %w", err)
+	}
+
+	// Create the initial post and save it.
 	startTime := time.Now()
 	post, postInfo, err := b.nipostBuilder.Proof(ctx, nodeID, shared.ZeroChallenge)
 	if err != nil {
-		return nil, nil, fmt.Errorf("post execution: %w", err)
+		return fmt.Errorf("post execution: %w", err)
 	}
 	if postInfo.Nonce == nil {
 		b.log.Error("initial PoST is invalid: missing VRF nonce. Check your PoST data",
 			log.ZShortStringer("smesherID", nodeID),
 		)
-		return nil, nil, errors.New("nil VRF nonce")
+		return errors.New("nil VRF nonce")
 	}
 	initialPost := nipost.Post{
 		Nonce:     post.Nonce,
@@ -349,127 +363,36 @@
 	}, postInfo.NumUnits)
 	if err != nil {
 		b.log.Error("initial POST is invalid", log.ZShortStringer("smesherID", nodeID), zap.Error(err))
-		return nil, nil, fmt.Errorf("initial POST is invalid: %w", err)
-	}
-
-	if err := nipost.AddPost(b.localDB, nodeID, initialPost); err != nil {
-		b.log.Error("failed to save initial post", 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)
 	}
 
 	metrics.PostDuration.Set(float64(time.Since(startTime).Nanoseconds()))
 	public.PostSeconds.Set(float64(time.Since(startTime)))
 	b.log.Info("created the initial post")
-	return post, postInfo, nil
-}
-
-// Obtain certificates for the poets.
-// We want to certify immediately after the startup or creating the initial POST
-// to avoid all nodes spamming the certifier at the same time when
-// submitting to the poets.
-func (b *Builder) certifyPost(ctx context.Context, nodeID types.NodeID, post *nipost.Post) {
-	client := NewCertifierClient(b.log, nodeID, post, WithCertifierClientConfig(b.certifierConfig.Client))
-	certifier := NewCertifier(b.localDB, b.log, client)
-	certifier.CertifyAll(ctx, b.poets)
 
-	b.smeshingMutex.Lock()
-	b.certifiers[nodeID] = certifier
-	b.smeshingMutex.Unlock()
+	return nipost.AddPost(b.localDB, nodeID, initialPost)
 }
 
-func (b *Builder) obtainPostFromLastAtx(ctx context.Context, nodeId types.NodeID) (*nipost.Post, error) {
-	atxid, err := atxs.GetLastIDByNodeID(b.db, nodeId)
-	if err != nil {
-		return nil, fmt.Errorf("no existing ATX found: %w", err)
-	}
-	atx, err := atxs.Get(b.db, atxid)
-	if err != nil {
-		return nil, fmt.Errorf("failed to retrieve ATX: %w", err)
-	}
-	if atx.NIPost == nil {
-		return nil, errors.New("no NIPoST found in last ATX")
-	}
-	if atx.CommitmentATX == nil {
-		if commitmentAtx, err := atxs.CommitmentATX(b.db, nodeId); err != nil {
-			return nil, fmt.Errorf("failed to retrieve commitment ATX: %w", err)
-		} else {
-			atx.CommitmentATX = &commitmentAtx
-		}
-	}
-	if atx.VRFNonce == nil {
-		if nonce, err := atxs.VRFNonce(b.db, nodeId, b.layerClock.CurrentLayer().GetEpoch()); err != nil {
-			return nil, fmt.Errorf("failed to retrieve VRF nonce: %w", err)
-		} else {
-			atx.VRFNonce = &nonce
-		}
-	}
-
-	b.log.Info("found POST in an existing ATX", zap.String("atx_id", atxid.Hash32().ShortString()))
-	return &nipost.Post{
-		Nonce:         atx.NIPost.Post.Nonce,
-		Indices:       atx.NIPost.Post.Indices,
-		Pow:           atx.NIPost.Post.Pow,
-		Challenge:     atx.NIPost.PostMetadata.Challenge,
-		NumUnits:      atx.NumUnits,
-		CommitmentATX: *atx.CommitmentATX,
-		VRFNonce:      *atx.VRFNonce,
-	}, nil
-}
-
-func (b *Builder) obtainPost(ctx context.Context, nodeID types.NodeID) (*nipost.Post, error) {
-	b.log.Info("looking for POST for poet certification")
-	post, err := nipost.GetPost(b.localDB, nodeID)
-	switch {
-	case err == nil:
-		b.log.Info("found POST in local DB")
-		return post, nil
-	case errors.Is(err, sql.ErrNotFound):
-		// no post found
-	default:
-		return nil, fmt.Errorf("loading initial post from db: %w", err)
-	}
-
-	b.log.Info("POST not found in local DB. Trying to obtain POST from an existing ATX")
-	if post, err := b.obtainPostFromLastAtx(ctx, nodeID); err == nil {
-		b.log.Info("found POST in an existing ATX")
-		if err := nipost.AddPost(b.localDB, nodeID, *post); err != nil {
-			b.log.Error("failed to save post", zap.Error(err))
-		}
-		return post, nil
-	}
+func (b *Builder) run(ctx context.Context, sig *signing.EdSigner) {
+	defer b.log.Info("atx builder stopped")
 
-	b.log.Info("POST not found in existing ATXs. Generating the initial POST")
 	for {
-		post, postInfo, err := b.buildInitialPost(ctx, nodeID)
+		err := b.buildInitialPost(ctx, sig.NodeID())
 		if err == nil {
-			return &nipost.Post{
-				Nonce:         post.Nonce,
-				Indices:       post.Indices,
-				Pow:           post.Pow,
-				Challenge:     shared.ZeroChallenge,
-				NumUnits:      postInfo.NumUnits,
-				CommitmentATX: postInfo.CommitmentATX,
-				VRFNonce:      *postInfo.Nonce,
-			}, nil
+			break
 		}
 		b.log.Error("failed to generate initial proof:", zap.Error(err))
 		currentLayer := b.layerClock.CurrentLayer()
 		select {
 		case <-ctx.Done():
-			return nil, ctx.Err()
+			return
 		case <-b.layerClock.AwaitLayer(currentLayer.Add(1)):
 		}
 	}
-}
-
-func (b *Builder) run(ctx context.Context, sig *signing.EdSigner) {
-	defer b.log.Info("atx builder stopped")
-
-	post, err := b.obtainPost(ctx, sig.NodeID())
-	if err != nil {
-		b.log.Error("failed to obtain post for certification", zap.Error(err))
-		return
-	}
-	b.certifyPost(ctx, sig.NodeID(), post)
+	b.certifier.CertifyAll(ctx, sig.NodeID(), b.poets)
 
 	for {
 		err := b.PublishActivationTx(ctx, sig)
@@ -616,7 +539,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)
@@ -746,17 +669,9 @@
 func (b *Builder) createAtx(
 	ctx context.Context,
 	sig *signing.EdSigner,
-	challenge *types.NIPostChallenge,
-) (*types.ActivationTx, error) {
-	pubEpoch := challenge.PublishEpoch
-	// TODO: in future, encode the right NiPoST challenge version depending on the pubEpoch.
-	challengeHash := wire.NIPostChallengeToWireV1(challenge).Hash()
-
-	b.smeshingMutex.Lock()
-	certifier := b.certifiers[sig.NodeID()]
-	b.smeshingMutex.Unlock()
-
-	nipostState, err := b.nipostBuilder.BuildNIPost(ctx, sig, challenge.PublishEpoch, challengeHash, certifier)
+	challenge *wire.NIPostChallengeV1,
+) (*wire.ActivationTxV1, error) {
+	nipostState, err := b.nipostBuilder.BuildNIPost(ctx, sig, challenge.PublishEpoch, challenge.Hash())
 	if err != nil {
 		return nil, fmt.Errorf("build NIPost: %w", err)
 	}

activation/activation_multi_test.go

@@ -250,11 +250,11 @@ func Test_Builder_Multi_InitialPost(t *testing.T) {
 				},
 				nil,
 			)
-			_, err := tab.obtainPost(context.Background(), sig.NodeID())
+			err := tab.buildInitialPost(context.Background(), sig.NodeID())
 			require.NoError(t, err)
 
 			// postClient.Proof() should not be called again
-			_, err = tab.obtainPost(context.Background(), sig.NodeID())
+			err = tab.buildInitialPost(context.Background(), sig.NodeID())
 			require.NoError(t, err)
 			return nil
 		})
@@ -397,7 +397,9 @@ func Test_Builder_Multi_HappyPath(t *testing.T) {
 			VRFNonce: types.VRFPostIndex(rand.Uint64()),
 		}
 		nipostState[sig.NodeID()] = state
-		tab.mnipost.EXPECT().BuildNIPost(gomock.Any(), sig, ref.Publish, ref.Hash(), gomock.Any()).Return(state, nil)
+		tab.mnipost.EXPECT().
+			BuildNIPost(gomock.Any(), sig, ref.PublishEpoch, ref.Hash()).
+			Return(state, nil)
 
 		// awaiting atx publication epoch log
 		tab.mclock.EXPECT().CurrentLayer().DoAndReturn(