nandi/glean · 2bc4161

nandi/gleanpublic⑂ Fork 0

⑂ 2bc4161

Tweaks in fetcher for improved fetchingUnverified

Julien Robert committed 2026-04-24T01:39:46+02:00 Browse files

2bc4161 parent: 9d8649d

modified internal/db/store.go +9 -5 Viewed

@@ -2,6 +2,7 @@ package db
2	2
3	3	import (
4	4	"context"
	5	+ "fmt"
5	6	"time"
6	7
7	8	"pkg.rbrt.fr/glean/internal/feed"
@@ -32,18 +33,21 @@ func (a *FeedStoreAdapter) RecordFetchError(ctx context.Context, feedURL, lastEr
32	33	}
33	34
34	35	func (a *FeedStoreAdapter) StoreFetchResult(ctx context.Context, feedURL string, articles []feed.Article, faviconURL string) error {
35		- if err := a.store.MarkFeedFetched(ctx, feedURL); err != nil {
36		- return err
37		- }
38	36	if len(articles) > 0 {
39	37	if err := a.store.UpsertArticlesBatch(ctx, articles); err != nil {
40		- return err
	38	+ return fmt.Errorf("failed to save articles: %w", err)
41	39	}
42	40	}
	41	+
	42	+ if err := a.store.MarkFeedFetched(ctx, feedURL); err != nil {
	43	+ return fmt.Errorf("failed to mark as fetched: %w", err)
	44	+ }
	45	+
43	46	if faviconURL != "" {
44	47	if err := a.store.UpdateFeedFavicon(ctx, feedURL, faviconURL); err != nil {
45		- return err
	48	+ return fmt.Errorf("failed to save favicon: %w", err)
46	49	}
47	50	}
	51	+
48	52	return nil
49	53	}

  import (
  	"context"
++	"fmt"
  	"time"
  	"pkg.rbrt.fr/glean/internal/feed"
  }
  func (a *FeedStoreAdapter) StoreFetchResult(ctx context.Context, feedURL string, articles []feed.Article, faviconURL string) error {
--	if err := a.store.MarkFeedFetched(ctx, feedURL); err != nil {
--		return err
--	}
  	if len(articles) > 0 {
  		if err := a.store.UpsertArticlesBatch(ctx, articles); err != nil {
--			return err
++			return fmt.Errorf("failed to save articles: %w", err)
  		}
  	}
++
++	if err := a.store.MarkFeedFetched(ctx, feedURL); err != nil {
++		return fmt.Errorf("failed to mark as fetched: %w", err)
++	}
++
  	if faviconURL != "" {
  		if err := a.store.UpdateFeedFavicon(ctx, feedURL, faviconURL); err != nil {
--			return err
++			return fmt.Errorf("failed to save favicon: %w", err)
  		}
  	}
++
  	return nil
  }

modified internal/feed/fetcher.go +9 -13 Viewed

@@ -49,6 +49,10 @@ func (f Fetcher) Fetch(ctx context.Context, feedURL string) (ParseResult, erro
49	49	}
50	50	}
51	51
	52	+ if lastResp != nil {
	53	+ lastResp.Body.Close()
	54	+ }
	55	+
52	56	result, resp, err := f.executeRequest(ctx, feedURL)
53	57	lastResp = resp
54	58	if err == nil {
@@ -80,18 +84,6 @@ func (f Fetcher) executeRequest(ctx context.Context, feedURL string) (ParseRes
80	84	}
81	85	defer resp.Body.Close()
82	86
83		- if resp.StatusCode == http.StatusTooManyRequests {
84		- return nil, resp, fmt.Errorf("rate limited (retry-after: %s)", resp.Header.Get("Retry-After"))
85		- }
86		-
87		- if resp.StatusCode >= 500 {
88		- return nil, resp, fmt.Errorf("server error: %d", resp.StatusCode)
89		- }
90		-
91		- if resp.StatusCode < 200 \|\| resp.StatusCode >= 300 {
92		- return nil, resp, fmt.Errorf("unexpected status: %d", resp.StatusCode)
93		- }
94		-
95	87	result, err := Parse(resp.Body, feedURL)
96	88	if err != nil {
97	89	return nil, nil, fmt.Errorf("parsing feed: %w", err)
@@ -159,12 +151,13 @@ func (s *Scheduler) Run(ctx context.Context) error {
159	151	}
160	152
161	153	func (s *Scheduler) fetchAll(ctx context.Context, olderThan time.Duration) {
162		- feeds, err := s.store.GetFeedsToFetch(ctx, olderThan, 10_000)
	154	+ feeds, err := s.store.GetFeedsToFetch(ctx, olderThan, 1000)
163	155	if err != nil {
164	156	s.logger.Error("failed to get feeds", "error", err)
165	157	return
166	158	}
167	159
	160	+ start := time.Now()
168	161	s.logger.Info("fetching feeds", "count", len(feeds), "older_than", olderThan)
169	162
170	163	g, gCtx := errgroup.WithContext(ctx)
@@ -176,11 +169,14 @@ func (s *Scheduler) fetchAll(ctx context.Context, olderThan time.Duration) {
176	169	})
177	170	}
178	171	_ = g.Wait()
	172	+
	173	+ s.logger.Info("fetching feeds complete", "duration", time.Since(start).Seconds())
179	174	}
180	175
181	176	func (s Scheduler) FetchFeed(ctx context.Context, feed Feed) {
182	177	call := &fetchCall{done: make(chan struct{})}
183	178	if actual, loaded := s.inFlight.LoadOrStore(feed.URL, call); loaded {
	179	+ s.logger.Debug("feed already in flight, skipping", "feed", feed.URL)
184	180	select {
185	181	case <-actual.(*fetchCall).done:
186	182	case <-ctx.Done():

  			}
  		}
++		if lastResp != nil {
++			lastResp.Body.Close()
++		}
++
  		result, resp, err := f.executeRequest(ctx, feedURL)
  		lastResp = resp
  		if err == nil {
  	}
  	defer resp.Body.Close()
--	if resp.StatusCode == http.StatusTooManyRequests {
--		return nil, resp, fmt.Errorf("rate limited (retry-after: %s)", resp.Header.Get("Retry-After"))
--	}
--
--	if resp.StatusCode >= 500 {
--		return nil, resp, fmt.Errorf("server error: %d", resp.StatusCode)
--	}
--
--	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
--		return nil, resp, fmt.Errorf("unexpected status: %d", resp.StatusCode)
--	}
--
  	result, err := Parse(resp.Body, feedURL)
  	if err != nil {
  		return nil, nil, fmt.Errorf("parsing feed: %w", err)
  }
  func (s *Scheduler) fetchAll(ctx context.Context, olderThan time.Duration) {
--	feeds, err := s.store.GetFeedsToFetch(ctx, olderThan, 10_000)
++	feeds, err := s.store.GetFeedsToFetch(ctx, olderThan, 1000)
  	if err != nil {
  		s.logger.Error("failed to get feeds", "error", err)
  		return
  	}
++	start := time.Now()
  	s.logger.Info("fetching feeds", "count", len(feeds), "older_than", olderThan)
  	g, gCtx := errgroup.WithContext(ctx)
  		})
  	}
  	_ = g.Wait()
++
++	s.logger.Info("fetching feeds complete", "duration", time.Since(start).Seconds())
  }
  func (s *Scheduler) FetchFeed(ctx context.Context, feed *Feed) {
  	call := &fetchCall{done: make(chan struct{})}
  	if actual, loaded := s.inFlight.LoadOrStore(feed.URL, call); loaded {
++		s.logger.Debug("feed already in flight, skipping", "feed", feed.URL)
  		select {
  		case <-actual.(*fetchCall).done:
  		case <-ctx.Done():