feat: add exponential backoff, some UI elements to see the resut of a scrape

scraper/internal/browser/cdp.go

@@ -15,6 +15,7 @@ import (
 // cdpClient implements BrowserClient using the CDP WebSocket endpoint.
 type cdpClient struct {
 	cfg Config
+	sem chan struct{}
 }
 
 // NewCDPClient returns a BrowserClient that uses CDP WebSocket sessions.
@@ -22,7 +23,7 @@ func NewCDPClient(cfg Config) BrowserClient {
 	if cfg.Timeout == 0 {
 		cfg.Timeout = 60 * time.Second
 	}
-	return &cdpClient{cfg: cfg}
+	return &cdpClient{cfg: cfg, sem: makeSem(cfg.MaxConcurrent)}
 }
 
 func (c *cdpClient) Strategy() Strategy { return StrategyCDP }
@@ -38,6 +39,11 @@ func (c *cdpClient) ScrapePage(_ context.Context, _ ScrapeRequest) (ScrapeRespon
 // CDPSession opens a WebSocket to the Browserless /devtools/browser endpoint,
 // navigates to pageURL, and invokes fn with a live CDPConn.
 func (c *cdpClient) CDPSession(ctx context.Context, pageURL string, fn CDPSessionFunc) error {
+	if err := acquire(ctx, c.sem); err != nil {
+		return fmt.Errorf("cdp: semaphore: %w", err)
+	}
+	defer release(c.sem)
+
 	// Build WebSocket URL: ws://host:port/devtools/browser?token=...&url=...
 	wsURL := strings.Replace(c.cfg.BaseURL, "http://", "ws://", 1)
 	wsURL = strings.Replace(wsURL, "https://", "wss://", 1)

scraper/internal/browser/content_scrape.go

@@ -18,12 +18,48 @@ type Config struct {
 	Token string
 	// Timeout is the per-request HTTP timeout; defaults to 60 s.
 	Timeout time.Duration
+	// MaxConcurrent caps the number of simultaneous in-flight requests sent to
+	// Browserless. When all slots are occupied new calls block until one
+	// completes (or ctx is cancelled). 0 means no limit.
+	MaxConcurrent int
+}
+
+// makeSem returns a buffered channel used as a counting semaphore.
+// If n <= 0 a nil channel is returned, which causes acquire/release to be no-ops.
+func makeSem(n int) chan struct{} {
+	if n <= 0 {
+		return nil
+	}
+	return make(chan struct{}, n)
+}
+
+// acquire takes one slot from sem. It returns an error if ctx is cancelled
+// before a slot becomes available. If sem is nil it returns immediately.
+func acquire(ctx context.Context, sem chan struct{}) error {
+	if sem == nil {
+		return nil
+	}
+	select {
+	case sem <- struct{}{}:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+// release frees the slot previously obtained by acquire.
+// If sem is nil it is a no-op.
+func release(sem chan struct{}) {
+	if sem != nil {
+		<-sem
+	}
 }
 
 // contentClient implements BrowserClient using the /content endpoint.
 type contentClient struct {
 	cfg  Config
 	http *http.Client
+	sem  chan struct{}
 }
 
 // NewContentClient returns a BrowserClient that uses POST /content.
@@ -34,12 +70,18 @@ func NewContentClient(cfg Config) BrowserClient {
 	return &contentClient{
 		cfg:  cfg,
 		http: &http.Client{Timeout: cfg.Timeout},
+		sem:  makeSem(cfg.MaxConcurrent),
 	}
 }
 
 func (c *contentClient) Strategy() Strategy { return StrategyContent }
 
 func (c *contentClient) GetContent(ctx context.Context, req ContentRequest) (string, error) {
+	if err := acquire(ctx, c.sem); err != nil {
+		return "", fmt.Errorf("content: semaphore: %w", err)
+	}
+	defer release(c.sem)
+
 	body, err := json.Marshal(req)
 	if err != nil {
 		return "", fmt.Errorf("content: marshal request: %w", err)
@@ -87,6 +129,7 @@ func (c *contentClient) CDPSession(_ context.Context, _ string, _ CDPSessionFunc
 type scrapeClient struct {
 	cfg  Config
 	http *http.Client
+	sem  chan struct{}
 }
 
 // NewScrapeClient returns a BrowserClient that uses POST /scrape.
@@ -97,6 +140,7 @@ func NewScrapeClient(cfg Config) BrowserClient {
 	return &scrapeClient{
 		cfg:  cfg,
 		http: &http.Client{Timeout: cfg.Timeout},
+		sem:  makeSem(cfg.MaxConcurrent),
 	}
 }
 
@@ -107,6 +151,11 @@ func (c *scrapeClient) GetContent(_ context.Context, _ ContentRequest) (string,
 }
 
 func (c *scrapeClient) ScrapePage(ctx context.Context, req ScrapeRequest) (ScrapeResponse, error) {
+	if err := acquire(ctx, c.sem); err != nil {
+		return ScrapeResponse{}, fmt.Errorf("scrape: semaphore: %w", err)
+	}
+	defer release(c.sem)
+
 	body, err := json.Marshal(req)
 	if err != nil {
 		return ScrapeResponse{}, fmt.Errorf("scrape: marshal request: %w", err)

scraper/internal/browser/interface.go

@@ -23,12 +23,18 @@ const (
 	StrategyCDP Strategy = "cdp"
 )
 
+// WaitForSelector describes the waitForSelector option sent to Browserless.
+type WaitForSelector struct {
+	Selector string `json:"selector"`
+	Timeout  int    `json:"timeout,omitempty"` // ms
+}
+
 // ContentRequest is the body sent to POST /content.
 type ContentRequest struct {
-	URL             string `json:"url"`
-	WaitFor         string `json:"waitForSelector,omitempty"`
-	WaitForTimeout  int    `json:"waitForTimeout,omitempty"` // ms
-	RejectResources bool   `json:"rejectResources,omitempty"`
+	URL                 string           `json:"url"`
+	WaitFor             *WaitForSelector `json:"waitForSelector,omitempty"`
+	WaitForTimeout      int              `json:"waitForTimeout,omitempty"`      // ms
+	RejectResourceTypes []string         `json:"rejectResourceTypes,omitempty"` // e.g. ["image","stylesheet"]
 }
 
 // ScrapeElement is one element descriptor inside a ScrapeRequest.
@@ -39,9 +45,9 @@ type ScrapeElement struct {
 
 // ScrapeRequest is the body sent to POST /scrape.
 type ScrapeRequest struct {
-	URL      string          `json:"url"`
-	Elements []ScrapeElement `json:"elements"`
-	WaitFor  string          `json:"waitForSelector,omitempty"`
+	URL      string           `json:"url"`
+	Elements []ScrapeElement  `json:"elements"`
+	WaitFor  *WaitForSelector `json:"waitForSelector,omitempty"`
 }
 
 // ScrapeResult is one entry in the response from POST /scrape.