feat(publisher): seperate generate ad and publish (#66)

seperate a parallizable step (generating ads) from one that is not
parallizable (publishing).

- create a generate ad function that just generates the add that will be
published for a given set of CIDs
- create an advertisement publisher that can batch several ads together
with only one head publish step
- port the generic queuepoller into go-libstoracha, now supporting
handlers that batch all jobs at once
- modify the queue interface to specifically seperate the job from the
ID on read
- rename queues to be called `publishing` rather than `publisher`
(consistency)
- modify extended queue to match queue interface
- add advertisement publishing SQS queue
- generate new handlers and interfaces for double step queue publishing

Author: hannahhoward

awsutils/extendedqueue.go

@@ -12,6 +12,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/aws/aws-sdk-go-v2/service/sqs"
 	"github.com/google/uuid"
+	"github.com/storacha/go-libstoracha/queuepoller"
 )
 
 // queueMessage is the struct that is serialized onto an SQS message queue in JSON
@@ -22,7 +23,6 @@ type queueMessage[Message any] struct {
 
 // SerializedJob represents a job that has been serialized for transport over SQS + S3
 type SerializedJob[Message any] struct {
-	ID       string
 	GroupID  *string
 	Message  Message
 	Extended io.Reader
@@ -31,7 +31,6 @@ type SerializedJob[Message any] struct {
 type JobMarshaller[Job any, Message any] interface {
 	Marshall(job Job) (SerializedJob[Message], error)
 	Unmarshall(SerializedJob[Message]) (Job, error)
-	Empty() Job
 }
 
 // SQSExtendedQueue implements a queue interface using SQS that can store extended data to an S3 bucket
@@ -115,7 +114,7 @@ func (s *SQSExtendedQueue[Job, Message]) sendMessage(ctx context.Context, groupI
 // Read reads a batch of jobs from the SQS queue.
 // Returns an empty slice if no jobs are available.
 // The caller must process jobs and delete them from the queue when done.
-func (s *SQSExtendedQueue[Job, Message]) Read(ctx context.Context, maxJobs int) ([]Job, error) {
+func (s *SQSExtendedQueue[Job, Message]) Read(ctx context.Context, maxJobs int) ([]queuepoller.WithID[Job], error) {
 	receiveOutput, err := s.sqsClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
 		QueueUrl:            aws.String(s.queueID),
 		MaxNumberOfMessages: int32(maxJobs),
@@ -126,10 +125,10 @@ func (s *SQSExtendedQueue[Job, Message]) Read(ctx context.Context, maxJobs int)
 	}
 
 	if len(receiveOutput.Messages) == 0 {
-		return []Job{}, nil
+		return []queuepoller.WithID[Job]{}, nil
 	}
 
-	jobs := make([]Job, 0, len(receiveOutput.Messages))
+	jobs := make([]queuepoller.WithID[Job], 0, len(receiveOutput.Messages))
 	for _, msg := range receiveOutput.Messages {
 		job, err := s.decoder.DecodeMessage(ctx, aws.ToString(msg.ReceiptHandle), aws.ToString(msg.Body))
 		if err != nil {
@@ -179,23 +178,26 @@ func NewSQSDecoder[Job any, Message any](cfg aws.Config, bucket string, marshall
 }
 
 // DecodeMessage decodes a provider caching job from the SQS message body, reading the stored index from S3
-func (s *SQSDecoder[Job, Message]) DecodeMessage(ctx context.Context, receiptHandle string, messageBody string) (Job, error) {
+func (s *SQSDecoder[Job, Message]) DecodeMessage(ctx context.Context, receiptHandle string, messageBody string) (queuepoller.WithID[Job], error) {
 	var msg queueMessage[Message]
 	err := json.Unmarshal([]byte(messageBody), &msg)
 	if err != nil {
-		return s.marshaller.Empty(), fmt.Errorf("deserializing message: %w", err)
+		return queuepoller.WithID[Job]{}, fmt.Errorf("deserializing message: %w", err)
 	}
 	received, err := s.s3Client.GetObject(ctx, &s3.GetObjectInput{
 		Bucket: aws.String(s.bucket),
 		Key:    aws.String(msg.JobID.String()),
 	})
 	if err != nil {
-		return s.marshaller.Empty(), fmt.Errorf("reading stored index CAR: %w", err)
+		return queuepoller.WithID[Job]{}, fmt.Errorf("reading stored index CAR: %w", err)
 	}
 	defer received.Body.Close()
-	return s.marshaller.Unmarshall(SerializedJob[Message]{
-		ID:       receiptHandle,
+	job, err := s.marshaller.Unmarshall(SerializedJob[Message]{
 		Message:  msg.Message,
 		Extended: received.Body,
 	})
+	if err != nil {
+		return queuepoller.WithID[Job]{}, fmt.Errorf("unmarshalling job: %w", err)
+	}
+	return queuepoller.WithID[Job]{ID: receiptHandle, Job: job}, nil
 }

go.mod

@@ -1,6 +1,6 @@
 module github.com/storacha/go-libstoracha
 
-go 1.24.4
+go 1.25.3
 
 require (
 	github.com/aws/aws-sdk-go v1.55.7

ipnipublisher/publisher/advertisementpublisher.go

@@ -0,0 +1,142 @@
+package publisher
+
+import (
+	"context"
+	"fmt"
+
+	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
+	"github.com/ipni/go-libipni/announce"
+	"github.com/ipni/go-libipni/announce/httpsender"
+	"github.com/ipni/go-libipni/dagsync/ipnisync/head"
+	"github.com/ipni/go-libipni/ingest/schema"
+	"github.com/libp2p/go-libp2p/core/crypto"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/storacha/go-libstoracha/ipnipublisher/store"
+	"github.com/storacha/go-ucanto/core/ipld"
+)
+
+type AdvertisementPublisher struct {
+	*options
+	pendingAds []schema.Advertisement
+	sender     announce.Sender
+	key        crypto.PrivKey
+	store      store.PublisherStore
+}
+
+func NewAdvertisementPublisher(id crypto.PrivKey, store store.PublisherStore, opts ...Option) (*AdvertisementPublisher, error) {
+	o := &options{
+		topic: "/indexer/ingest/mainnet",
+	}
+	for _, opt := range opts {
+		err := opt(o)
+		if err != nil {
+			return nil, err
+		}
+	}
+	peer, err := peer.IDFromPrivateKey(id)
+	if err != nil {
+		return nil, fmt.Errorf("cannot get peer ID from private key: %w", err)
+	}
+	batchPublisher := &AdvertisementPublisher{
+		options: o,
+		key:     id,
+		store:   store,
+	}
+	if len(o.announceURLs) > 0 {
+		sender, err := httpsender.New(o.announceURLs, peer)
+		if err != nil {
+			return nil, fmt.Errorf("cannot create http announce sender: %w", err)
+		}
+		log.Info("HTTP announcements enabled")
+		batchPublisher.sender = sender
+	}
+	return batchPublisher, nil
+}
+
+func (p *AdvertisementPublisher) AddToBatch(adv schema.Advertisement) error {
+	p.pendingAds = append(p.pendingAds, adv)
+	return nil
+}
+
+func (p *AdvertisementPublisher) Commit(ctx context.Context) (ipld.Link, error) {
+	pendingAds := p.pendingAds
+	p.pendingAds = nil
+	lnk, err := p.commit(ctx, pendingAds)
+	if err != nil {
+		for _, adv := range pendingAds {
+			if !adv.IsRm {
+				peer, err := peer.Decode(adv.Provider)
+				if err == nil {
+					_ = p.store.DeleteChunkLinkForProviderAndContextID(ctx, peer, adv.ContextID)
+				}
+			}
+		}
+		return nil, err
+	}
+	return lnk, nil
+}
+func (p *AdvertisementPublisher) commit(ctx context.Context, pendingAds []schema.Advertisement) (ipld.Link, error) {
+
+	// Get the previous advertisement that was generated.
+	prevHead, err := p.store.Head(ctx)
+	if err != nil {
+		if !store.IsNotFound(err) {
+			return nil, fmt.Errorf("could not get latest advertisement: %s", err)
+		}
+	}
+	var prevLink ipld.Link
+	// Check for cid.Undef for the previous link. If this is the case, then
+	// this means there are no previous advertisements.
+	if prevHead == nil {
+		log.Info("Latest advertisement CID was undefined - no previous advertisement")
+	} else {
+		prevLink = prevHead.Head
+	}
+
+	if len(pendingAds) == 0 {
+		log.Info("No pending advertisements to commit")
+		return prevLink, nil
+	}
+
+	// Store all pending advertisements in order, linking each to the previous.
+	for _, adv := range pendingAds {
+		adv.PreviousID = prevLink
+
+		// Sign the advertisement.
+		if err = adv.Sign(p.key); err != nil {
+			return nil, err
+		}
+
+		if err := adv.Validate(); err != nil {
+			return nil, err
+		}
+
+		lnk, err := p.store.PutAdvert(ctx, adv)
+		if err != nil {
+			return nil, err
+		}
+		log.Info("Stored ad in local link system")
+		prevLink = lnk
+	}
+
+	lnk := prevLink
+	head, err := head.NewSignedHead(lnk.(cidlink.Link).Cid, p.topic, p.key)
+	if err != nil {
+		log.Errorw("Failed to generate signed head for the latest advertisement", "err", err)
+		return nil, fmt.Errorf("failed to generate signed head for the latest advertisement: %w", err)
+	}
+	if _, err := p.store.ReplaceHead(ctx, prevHead, head); err != nil {
+		log.Errorw("Failed to update reference to the latest advertisement", "err", err)
+		return nil, fmt.Errorf("failed to update reference to latest advertisement: %w", err)
+	}
+	log.Info("Updated reference to the latest advertisement successfully")
+
+	if p.sender != nil {
+		err = announce.Send(ctx, lnk.(cidlink.Link).Cid, p.pubHTTPAnnounceAddrs, p.sender)
+		if err != nil {
+			log.Warnw("Failed to announce advertisement", "err", err)
+		}
+	}
+
+	return lnk, nil
+}

ipnipublisher/publisher/generate_ad.go

@@ -0,0 +1,125 @@
+package publisher
+
+import (
+	"context"
+	"encoding/base64"
+	"fmt"
+	"iter"
+
+	"github.com/ipni/go-libipni/ingest/schema"
+	"github.com/ipni/go-libipni/metadata"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/multiformats/go-multiaddr"
+	mh "github.com/multiformats/go-multihash"
+	"github.com/storacha/go-libstoracha/ipnipublisher/store"
+)
+
+// GenerateAd generates an advertisement for the given parameters.
+func GenerateAd(ctx context.Context, publisherStore store.PublisherStore, peer peer.ID, addrs []multiaddr.Multiaddr, contextID []byte, md metadata.Metadata, isRm bool, mhs iter.Seq[mh.Multihash]) (schema.Advertisement, error) {
+	var err error
+
+	log := log.With("providerID", peer).With("contextID", base64.StdEncoding.EncodeToString(contextID))
+
+	chunkLink, err := publisherStore.ChunkLinkForProviderAndContextID(ctx, peer, contextID)
+	if err != nil {
+		if !store.IsNotFound(err) {
+			return schema.Advertisement{}, fmt.Errorf("could not get entries cid by provider + context id: %s", err)
+		}
+	}
+
+	// If not removing, then generate the link for the list of CIDs from the
+	// contextID using the multihash lister, and store the relationship.
+	if !isRm {
+		log.Info("Creating advertisement")
+
+		// If no previously-published ad for this context ID.
+		if chunkLink == nil {
+			log.Info("Generating entries linked list for advertisement")
+
+			// Generate the linked list ipld.Link that is added to the
+			// advertisement and used for ingestion.
+			chunkLink, err = publisherStore.PutEntries(ctx, mhs)
+			if err != nil {
+				return schema.Advertisement{}, fmt.Errorf("could not generate entries list: %s", err)
+			}
+			if chunkLink == nil {
+				log.Warnw("chunking for context ID resulted in no link", "contextID", contextID)
+				chunkLink = schema.NoEntries
+			}
+
+			// Store the relationship between providerID, contextID and CID of the
+			// advertised list of Cids.
+			err = publisherStore.PutChunkLinkForProviderAndContextID(ctx, peer, contextID, chunkLink)
+			if err != nil {
+				return schema.Advertisement{}, fmt.Errorf("failed to write provider + context id to entries cid mapping: %s", err)
+			}
+		} else {
+			// Lookup metadata for this providerID and contextID.
+			prevMetadata, err := publisherStore.MetadataForProviderAndContextID(ctx, peer, contextID)
+			if err != nil {
+				if !store.IsNotFound(err) {
+					return schema.Advertisement{}, fmt.Errorf("could not get metadata for provider + context id: %s", err)
+				}
+				log.Warn("No metadata for existing provider + context ID, generating new advertisement")
+			}
+
+			if md.Equal(prevMetadata) {
+				// Metadata is the same; no change, no need for new
+				// advertisement.
+				return schema.Advertisement{}, ErrAlreadyAdvertised
+			}
+
+			// Linked list is the same, but metadata is different, so generate
+			// new advertisement with same linked list, but new metadata.
+		}
+
+		if err = publisherStore.PutMetadataForProviderAndContextID(ctx, peer, contextID, md); err != nil {
+			return schema.Advertisement{}, fmt.Errorf("failed to write provider + context id to metadata mapping: %s", err)
+		}
+	} else {
+		log.Info("Creating removal advertisement")
+
+		if chunkLink == nil {
+			return schema.Advertisement{}, ErrContextIDNotFound
+		}
+
+		// If removing by context ID, it means the list of CIDs is not needed
+		// anymore, so we can remove the entry from the datastore.
+		err = publisherStore.DeleteChunkLinkForProviderAndContextID(ctx, peer, contextID)
+		if err != nil {
+			return schema.Advertisement{}, fmt.Errorf("failed to delete provider + context id to entries cid mapping: %s", err)
+		}
+		err = publisherStore.DeleteMetadataForProviderAndContextID(ctx, peer, contextID)
+		if err != nil {
+			return schema.Advertisement{}, fmt.Errorf("failed to delete provider + context id to metadata mapping: %s", err)
+		}
+
+		// Create an advertisement to delete content by contextID by specifying
+		// that advertisement has no entries.
+		chunkLink = schema.NoEntries
+
+		// The advertisement still requires a valid metadata even though
+		// metadata is not used for removal. Create a valid empty metadata.
+		md = metadata.Default.New()
+	}
+
+	mdBytes, err := md.MarshalBinary()
+	if err != nil {
+		return schema.Advertisement{}, err
+	}
+
+	var stringAddrs []string
+	for _, addr := range addrs {
+		stringAddrs = append(stringAddrs, addr.String())
+	}
+
+	return schema.Advertisement{
+		Provider:  peer.String(),
+		Addresses: stringAddrs,
+		Entries:   chunkLink,
+		ContextID: contextID,
+		Metadata:  mdBytes,
+		IsRm:      isRm,
+	}, nil
+
+}