Rewriting the queue

.github/workflows/go.yml

@@ -22,7 +22,6 @@ jobs:
       run: go build -v ./...
 
     - name: Test
-      continue-on-error: true
       run: go test -race -v ./...
 
     - name: Goroutine leak detector

cmd/get/list.go

@@ -3,7 +3,6 @@ package get
 import (
 	"github.com/internetarchive/Zeno/cmd"
 	"github.com/internetarchive/Zeno/config"
-	"github.com/internetarchive/Zeno/internal/pkg/frontier"
 	"github.com/sirupsen/logrus"
 	log "github.com/sirupsen/logrus"
 	"github.com/urfave/cli/v2"
@@ -30,7 +29,7 @@ func cmdGetList(c *cli.Context) error {
 	crawl := cmd.InitCrawlWithCMD(config.App.Flags)
 
 	// Initialize initial seed list
-	crawl.SeedList, err = frontier.IsSeedList(c.Args().Get(0))
+	crawl.SeedList, err = isSeedList(c.Args().Get(0))
 	if err != nil || len(crawl.SeedList) <= 0 {
 		logrus.WithFields(logrus.Fields{
 			"input": c.Args().Get(0),

cmd/get/url.go

@@ -5,7 +5,7 @@ import (
 
 	"github.com/internetarchive/Zeno/cmd"
 	"github.com/internetarchive/Zeno/config"
-	"github.com/internetarchive/Zeno/internal/pkg/frontier"
+	"github.com/internetarchive/Zeno/internal/pkg/queue"
 	"github.com/sirupsen/logrus"
 	"github.com/urfave/cli/v2"
 )
@@ -40,7 +40,7 @@ func cmdGetURL(c *cli.Context) error {
 		return err
 	}
 
-	crawl.SeedList = append(crawl.SeedList, *frontier.NewItem(input, nil, "seed", 0, "", false))
+	crawl.SeedList = append(crawl.SeedList, *queue.NewItem(input, nil, "seed", 0, "", false))
 
 	// Start crawl
 	err = crawl.Start()

internal/pkg/frontier/utils.go → cmd/get/utils.go

@@ -1,21 +1,18 @@
-package frontier
+package get
 
 import (
 	"bufio"
-	"encoding/gob"
 	"errors"
 	"fmt"
 	"net/url"
 	"os"
-	"sync"
 
 	"github.com/gosuri/uilive"
+	"github.com/internetarchive/Zeno/internal/pkg/queue"
 	"github.com/sirupsen/logrus"
 )
 
-// IsSeedList validates if the path is a seed list, and return an array of
-// frontier.Item made of the seeds if it can
-func IsSeedList(path string) (seeds []Item, err error) {
+func isSeedList(path string) (seeds []queue.Item, err error) {
 	var totalCount, validCount int
 	writer := uilive.New()
 	writer.Start()
@@ -51,7 +48,7 @@ func IsSeedList(path string) (seeds []Item, err error) {
 			continue
 		}
 
-		item := NewItem(URL, nil, "seed", 0, "", false)
+		item := queue.NewItem(URL, nil, "seed", 0, "", false)
 		seeds = append(seeds, *item)
 		validCount++
 		fmt.Fprintf(writer, "\t   Reading input list.. Found %d valid URLs out of %d URLs read.\n", validCount, totalCount)
@@ -69,40 +66,3 @@ func IsSeedList(path string) (seeds []Item, err error) {
 
 	return seeds, nil
 }
-
-type Pair struct {
-	Key, Value interface{}
-}
-
-func SyncMapEncode(m *sync.Map, file *os.File) error {
-	var pairs []Pair
-
-	m.Range(func(key, value interface{}) bool {
-		pairs = append(pairs, Pair{key, value})
-		return true
-	})
-
-	gob.Register(PoolItem{})
-
-	enc := gob.NewEncoder(file)
-	err := enc.Encode(pairs)
-
-	return err
-}
-
-func SyncMapDecode(m *sync.Map, file *os.File) error {
-	var pairs []Pair
-	gob.Register(PoolItem{})
-	dec := gob.NewDecoder(file)
-	err := dec.Decode(&pairs)
-
-	if err != nil {
-		return err
-	}
-
-	for _, p := range pairs {
-		m.Store(p.Key, p.Value.(PoolItem))
-	}
-
-	return nil
-}

cmd/utils.go

@@ -12,7 +12,6 @@ import (
 	"github.com/google/uuid"
 	"github.com/internetarchive/Zeno/config"
 	"github.com/internetarchive/Zeno/internal/pkg/crawl"
-	"github.com/internetarchive/Zeno/internal/pkg/frontier"
 	"github.com/internetarchive/Zeno/internal/pkg/log"
 	"github.com/internetarchive/Zeno/internal/pkg/utils"
 	"github.com/paulbellamy/ratecounter"
@@ -73,10 +72,6 @@ func InitCrawlWithCMD(flags config.Flags) *crawl.Crawl {
 
 	c.LiveStats = flags.LiveStats
 
-	// Frontier
-	c.Frontier = new(frontier.Frontier)
-	c.Frontier.Log = c.Log
-
 	// If the job name isn't specified, we generate a random name
 	if flags.Job == "" {
 		if flags.HQProject != "" {
@@ -101,7 +96,7 @@ func InitCrawlWithCMD(flags config.Flags) *crawl.Crawl {
 	c.MaxConcurrentAssets = flags.MaxConcurrentAssets
 	c.WorkerStopSignal = make(chan bool)
 
-	c.Seencheck = flags.Seencheck
+	c.UseSeencheck = flags.Seencheck
 	c.HTTPTimeout = flags.HTTPTimeout
 	c.MaxConcurrentRequestsPerDomain = flags.MaxConcurrentRequestsPerDomain
 	c.RateLimitDelay = flags.RateLimitDelay

internal/pkg/crawl/api.go

@@ -43,7 +43,7 @@ func (crawl *Crawl) startAPI() {
 			"crawled":       crawledSeeds + crawledAssets,
 			"crawledSeeds":  crawledSeeds,
 			"crawledAssets": crawledAssets,
-			"queued":        crawl.Frontier.QueueCount.Value(),
+			"queued":        crawl.Queue.GetStats().TotalElements,
 			"uptime":        time.Since(crawl.StartTime).String(),
 		}
 
@@ -52,6 +52,10 @@ func (crawl *Crawl) startAPI() {
 
 	http.HandleFunc("/metrics", setupPrometheus(crawl).ServeHTTP)
 
+	http.HandleFunc("/queue", func(w http.ResponseWriter, r *http.Request) {
+		json.NewEncoder(w).Encode(crawl.Queue.GetStats())
+	})
+
 	http.HandleFunc("/workers", func(w http.ResponseWriter, r *http.Request) {
 		workersState := crawl.GetWorkerState(-1)
 		json.NewEncoder(w).Encode(workersState)

internal/pkg/crawl/assets.go

@@ -7,11 +7,11 @@ import (
 
 	"github.com/PuerkitoBio/goquery"
 	"github.com/internetarchive/Zeno/internal/pkg/crawl/sitespecific/cloudflarestream"
-	"github.com/internetarchive/Zeno/internal/pkg/frontier"
+	"github.com/internetarchive/Zeno/internal/pkg/queue"
 	"github.com/internetarchive/Zeno/internal/pkg/utils"
 )
 
-func (c *Crawl) extractAssets(base *url.URL, item *frontier.Item, doc *goquery.Document) (assets []*url.URL, err error) {
+func (c *Crawl) extractAssets(base *url.URL, item *queue.Item, doc *goquery.Document) (assets []*url.URL, err error) {
 	var rawAssets []string
 
 	// Execute plugins on the response

internal/pkg/crawl/capture.go

@@ -20,23 +20,24 @@ import (
 	"github.com/internetarchive/Zeno/internal/pkg/crawl/sitespecific/tiktok"
 	"github.com/internetarchive/Zeno/internal/pkg/crawl/sitespecific/truthsocial"
 	"github.com/internetarchive/Zeno/internal/pkg/crawl/sitespecific/vk"
+	"github.com/internetarchive/Zeno/internal/pkg/queue"
 	"github.com/internetarchive/Zeno/internal/pkg/utils"
 	"github.com/remeh/sizedwaitgroup"
 	"github.com/tomnomnom/linkheader"
-
-	"github.com/internetarchive/Zeno/internal/pkg/frontier"
 )
 
-func (c *Crawl) executeGET(item *frontier.Item, req *http.Request, isRedirection bool) (resp *http.Response, err error) {
+func (c *Crawl) executeGET(item *queue.Item, req *http.Request, isRedirection bool) (resp *http.Response, err error) {
 	var (
 		executionStart = time.Now()
-		newItem        *frontier.Item
+		newItem        *queue.Item
 		newReq         *http.Request
 		URL            *url.URL
 	)
 
 	defer func() {
-		c.PrometheusMetrics.DownloadedURI.Inc()
+		if c.API {
+			c.PrometheusMetrics.DownloadedURI.Inc()
+		}
 
 		c.URIsPerSecond.Incr(1)
 
@@ -52,17 +53,17 @@ func (c *Crawl) executeGET(item *frontier.Item, req *http.Request, isRedirection
 		time.Sleep(time.Second)
 	}
 
+	// TODO: re-implement host limitation
 	// Temporarily pause crawls for individual hosts if they are over our configured maximum concurrent requests per domain.
 	// If the request is a redirection, we do not pause the crawl because we want to follow the redirection.
-	if !isRedirection {
-		for c.shouldPause(item.Host) {
-			time.Sleep(time.Millisecond * time.Duration(c.RateLimitDelay))
-		}
+	// if !isRedirection {
+	// for c.shouldPause(item.Host) {
+	// 	time.Sleep(time.Millisecond * time.Duration(c.RateLimitDelay))
+	// }
 
-		c.Frontier.IncrHostActive(item.Host)
-
-		defer c.Frontier.DecrHostActive(item.Host)
-	}
+	// c.Queue.IncrHostActive(item.Host)
+	// defer c.Frontier.DecrHostActive(item.Host)
+	//}
 
 	// Retry on 429 error
 	for retry := 0; retry < c.MaxRetry; retry++ {
@@ -149,7 +150,7 @@ func (c *Crawl) executeGET(item *frontier.Item, req *http.Request, isRedirection
 		}
 
 		// Seencheck the URL
-		if c.Seencheck {
+		if c.UseSeencheck {
 			found := c.seencheckURL(utils.URLToString(URL), "seed")
 			if found {
 				return nil, errors.New("URL from redirection has already been seen")
@@ -165,7 +166,7 @@ func (c *Crawl) executeGET(item *frontier.Item, req *http.Request, isRedirection
 			}
 		}
 
-		newItem = frontier.NewItem(URL, item, item.Type, item.Hop, item.ID, false)
+		newItem = queue.NewItem(URL, item, item.Type, item.Hop, item.ID, false)
 		newItem.Redirect = item.Redirect + 1
 
 		// Prepare GET request
@@ -184,7 +185,7 @@ func (c *Crawl) executeGET(item *frontier.Item, req *http.Request, isRedirection
 	return resp, nil
 }
 
-func (c *Crawl) captureAsset(item *frontier.Item, cookies []*http.Cookie) error {
+func (c *Crawl) captureAsset(item *queue.Item, cookies []*http.Cookie) error {
 	var resp *http.Response
 
 	// Prepare GET request
@@ -216,13 +217,13 @@ func (c *Crawl) captureAsset(item *frontier.Item, cookies []*http.Cookie) error
 }
 
 // Capture capture the URL and return the outlinks
-func (c *Crawl) Capture(item *frontier.Item) error {
+func (c *Crawl) Capture(item *queue.Item) error {
 	var (
 		resp      *http.Response
 		waitGroup sync.WaitGroup
 	)
 
-	defer func(i *frontier.Item) {
+	defer func(i *queue.Item) {
 		waitGroup.Wait()
 
 		if c.UseHQ && i.ID != "" {
@@ -254,7 +255,7 @@ func (c *Crawl) Capture(item *frontier.Item) error {
 				c.Log.WithFields(c.genLogFields(err, item.URL, nil)).Error("error while generating API URL")
 			} else {
 				// Then we create an item
-				apiItem := frontier.NewItem(apiURL, item, item.Type, item.Hop, item.ID, false)
+				apiItem := queue.NewItem(apiURL, item, item.Type, item.Hop, item.ID, false)
 
 				// And capture it
 				c.Capture(apiItem)
@@ -267,7 +268,7 @@ func (c *Crawl) Capture(item *frontier.Item) error {
 			} else {
 				for _, embedURL := range embedURLs {
 					// Create the embed item
-					embedItem := frontier.NewItem(embedURL, item, item.Type, item.Hop, item.ID, false)
+					embedItem := queue.NewItem(embedURL, item, item.Type, item.Hop, item.ID, false)
 
 					// Capture the embed item
 					c.Capture(embedItem)
@@ -284,7 +285,7 @@ func (c *Crawl) Capture(item *frontier.Item) error {
 				c.Log.WithFields(c.genLogFields(err, item.URL, nil)).Error("error while generating Facebook embed URL")
 			} else {
 				// Create the embed item
-				embedItem := frontier.NewItem(embedURL, item, item.Type, item.Hop, item.ID, false)
+				embedItem := queue.NewItem(embedURL, item, item.Type, item.Hop, item.ID, false)
 
 				// Capture the embed item
 				err = c.Capture(embedItem)
@@ -302,7 +303,7 @@ func (c *Crawl) Capture(item *frontier.Item) error {
 			if highwindsURL == nil {
 				c.Log.WithFields(c.genLogFields(err, item.URL, nil)).Error("error while generating libsyn URL")
 			} else {
-				c.Capture(frontier.NewItem(highwindsURL, item, item.Type, item.Hop, item.ID, false))
+				c.Capture(queue.NewItem(highwindsURL, item, item.Type, item.Hop, item.ID, false))
 			}
 		}
 	} else if tiktok.IsTikTokURL(utils.URLToString(item.URL)) {
@@ -312,7 +313,7 @@ func (c *Crawl) Capture(item *frontier.Item) error {
 		telegram.TransformURL(item.URL)
 
 		// Then we create an item
-		embedItem := frontier.NewItem(item.URL, item, item.Type, item.Hop, item.ID, false)
+		embedItem := queue.NewItem(item.URL, item, item.Type, item.Hop, item.ID, false)
 
 		// And capture it
 		c.Capture(embedItem)
@@ -325,7 +326,7 @@ func (c *Crawl) Capture(item *frontier.Item) error {
 	if err != nil && err.Error() == "URL from redirection has already been seen" {
 		return err
 	} else if err != nil && err.Error() == "URL is being rate limited, sending back to HQ" {
-		c.HQProducerChannel <- frontier.NewItem(item.URL, item.ParentItem, item.Type, item.Hop, "", true)
+		c.HQProducerChannel <- queue.NewItem(item.URL, item.ParentItem, item.Type, item.Hop, "", true)
 		c.Log.WithFields(c.genLogFields(err, item.URL, nil)).Error("URL is being rate limited, sending back to HQ")
 		return err
 	} else if err != nil {
@@ -428,7 +429,7 @@ func (c *Crawl) Capture(item *frontier.Item) error {
 		// Seencheck the URLs we captured, we ignore the returned value here
 		// because we already archived the URLs, we just want them to be added
 		// to the seencheck table.
-		if c.Seencheck {
+		if c.UseSeencheck {
 			for _, cfstreamURL := range cfstreamURLs {
 				c.seencheckURL(cfstreamURL, "asset")
 			}
@@ -505,7 +506,7 @@ func (c *Crawl) Capture(item *frontier.Item) error {
 	// If --local-seencheck is enabled, then we check if the assets are in the
 	// seencheck DB. If they are, then they are skipped.
 	// Else, if we use HQ, then we use HQ's seencheck.
-	if c.Seencheck {
+	if c.UseSeencheck {
 		seencheckedBatch := []*url.URL{}
 
 		for _, URL := range assets {
@@ -541,12 +542,16 @@ func (c *Crawl) Capture(item *frontier.Item) error {
 		}
 	}
 
-	c.Frontier.QueueCount.Incr(int64(len(assets)))
+	// TODO: implement a counter for the number of assets
+	// currently being processed
+	// c.Frontier.QueueCount.Incr(int64(len(assets)))
 	swg := sizedwaitgroup.New(c.MaxConcurrentAssets)
 	excluded := false
 
 	for _, asset := range assets {
-		c.Frontier.QueueCount.Incr(-1)
+		// TODO: implement a counter for the number of assets
+		// currently being processed
+		// c.Frontier.QueueCount.Incr(-1)
 
 		// Just making sure we do not over archive by archiving the original URL
 		if utils.URLToString(item.URL) == utils.URLToString(asset) {
@@ -579,7 +584,7 @@ func (c *Crawl) Capture(item *frontier.Item) error {
 			defer swg.Done()
 
 			// Create the asset's item
-			newAsset := frontier.NewItem(asset, item, "asset", item.Hop, "", false)
+			newAsset := queue.NewItem(asset, item, "asset", item.Hop, "", false)
 
 			// Capture the asset
 			err = c.captureAsset(newAsset, resp.Cookies())