16810397243

Committed 07 Aug 2025 04:38PM UTC coverage: 74.443% (+0.09%) from 74.356%

Build # 16810397243

Build Type

push

github

Committed by

umputun

Commit Message

Add comprehensive logging for Telegram timeout debugging

- Add detailed logging for each retry attempt with timing information
- Log file sizes, titles, and elapsed time for each phase
- Separate logging for download vs copy operations
- Track attempt numbers to understand retry behavior
- This will help diagnose timeout issues with large audio files

Coverage Stats

19 of 27 new or added lines in 2 files covered. (70.37%)

1 existing line in 1 file now uncovered.

1404 of 1886 relevant lines covered (74.44%)

8.7 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

75.31

/app/proc/processor.go

// Package proc provided the primary blocking loop
// updating from sources and making feeds
package proc

import (
        "context"
        "time"

        log "github.com/go-pkgz/lgr"
        "github.com/go-pkgz/repeater"
        "github.com/go-pkgz/syncs"

        "github.com/umputun/feed-master/app/config"
        "github.com/umputun/feed-master/app/feed"
)

//go:generate moq -out mocks/telegram_notif.go -pkg mocks -skip-ensure -fmt goimports . TelegramNotif
//go:generate moq -out mocks/twitter_notif.go -pkg mocks -skip-ensure -fmt goimports . TwitterNotif

// TelegramNotif is interface to send messages to telegram
type TelegramNotif interface {
        Send(chanID string, item feed.Item) error
}

// TwitterNotif is interface to send message to twitter
type TwitterNotif interface {
        Send(item feed.Item) error
}

// Processor is a feed reader and store writer
type Processor struct {
        Conf          *config.Conf
        Store         *BoltDB
        TelegramNotif TelegramNotif
        TwitterNotif  TwitterNotif
}

// Do activate loop of goroutine for each feed, concurrency limited by p.Conf.Concurrent
func (p *Processor) Do(ctx context.Context) error {
        log.Printf("[INFO] activate processor, feeds=%d, %+v", len(p.Conf.Feeds), p.Conf)

        for {
                select {
                case <-ctx.Done():
                        return ctx.Err()
                default:
                        p.processFeeds(ctx)
                }
        }
}

func (p *Processor) processFeeds(ctx context.Context) {
        log.Printf("[DEBUG] refresh started")
        swg := syncs.NewSizedGroup(p.Conf.System.Concurrent, syncs.Preemptive, syncs.Context(ctx))
        for name, fm := range p.Conf.Feeds {
                for _, src := range fm.Sources {
                        name, src, fm := name, src, fm
                        swg.Go(func(context.Context) {
                                p.processFeed(name, src.URL, fm.TelegramChannel, p.Conf.System.MaxItems, fm.Filter)
                        })
                }
        }
        swg.Wait()
        log.Printf("[DEBUG] refresh completed")
        time.Sleep(p.Conf.System.UpdateInterval)
}

func (p *Processor) processFeed(name, url, telegramChannel string, maximum int, filter config.Filter) {
        rss, err := feed.Parse(url)
        if err != nil {
                log.Printf("[WARN] failed to parse %s, %v", url, err)
                return
        }

        // up to MaxItems (5) items from each feed
        upto := maximum
        if len(rss.ItemList) <= maximum {
                upto = len(rss.ItemList)
        }

        for _, item := range rss.ItemList[:upto] {
                // skip 1y and older
                if item.DT.Before(time.Now().AddDate(-1, 0, 0)) {
                        continue
                }

                skip, err := filter.Skip(item)
                if err != nil {
                        log.Printf("[WARN] failed to filter %s (%s) to %s, save as is, %v", item.GUID, item.PubDate, name, err)
                }
                if skip {
                        item.Junk = true
                        log.Printf("[INFO] filtered %s (%s), %s %s", item.GUID, item.PubDate, name, item.Title)
                }

                created, err := p.Store.Save(name, item)
                if err != nil {
                        log.Printf("[WARN] failed to save %s (%s) to %s, %v", item.GUID, item.PubDate, name, err)
                }

                // don't attempt to send anything if the entry was already saved
                // or in case it was filtered out
                if !created || item.Junk {
                        continue
                }

                rptr := repeater.NewDefault(3, 5*time.Second)
                attemptNum := 0
                err = rptr.Do(context.Background(), func() error {
                        attemptNum++
                        startTime := time.Now()
                        log.Printf("[DEBUG] sending telegram message (attempt %d/3): title=%q, size=%d bytes, url=%s to channel=%s",
                                attemptNum, item.Title, item.Enclosure.Length, item.Enclosure.URL, telegramChannel)
                        
                        if e := p.TelegramNotif.Send(telegramChannel, item); e != nil {
                                elapsed := time.Since(startTime)
                                log.Printf("[WARN] failed attempt %d/3 to send telegram message after %v: title=%q, size=%d bytes, url=%s to channel=%s, error=%v",
                                        attemptNum, elapsed, item.Title, item.Enclosure.Length, item.Enclosure.URL, telegramChannel, e)
                                return e
                        }
                        
                        elapsed := time.Since(startTime)
                        log.Printf("[INFO] successfully sent telegram message in %v: title=%q, size=%d bytes",
                                elapsed, item.Title, item.Enclosure.Length)
                        return nil
                })
                if err != nil {
                        log.Printf("[WARN] failed to send telegram message after 3 attempts: title=%q, size=%d bytes, url=%s to channel=%s, final_error=%v",
                                item.Title, item.Enclosure.Length, item.Enclosure.URL, telegramChannel, err)
                }

                if err := p.TwitterNotif.Send(item); err != nil {
                        log.Printf("[WARN] failed send twitter message, url=%s, %v", item.Enclosure.URL, err)
                }
        }

        // keep up to MaxKeepInDB items in bucket
        if removed, err := p.Store.removeOld(name, p.Conf.System.MaxKeepInDB); err == nil {
                if removed > 0 {
                        log.Printf("[DEBUG] removed %d from %s", removed, name)
                }
        } else {
                log.Printf("[WARN] failed to remove, %v", err)
        }
}

1	// Package proc provided the primary blocking loop
2	// updating from sources and making feeds
3	package proc
4
5	import (
6	"context"
7	"time"
8
9	log "github.com/go-pkgz/lgr"
10	"github.com/go-pkgz/repeater"
11	"github.com/go-pkgz/syncs"
12
13	"github.com/umputun/feed-master/app/config"
14	"github.com/umputun/feed-master/app/feed"
15	)
16
17	//go:generate moq -out mocks/telegram_notif.go -pkg mocks -skip-ensure -fmt goimports . TelegramNotif
18	//go:generate moq -out mocks/twitter_notif.go -pkg mocks -skip-ensure -fmt goimports . TwitterNotif
19
20	// TelegramNotif is interface to send messages to telegram
21	type TelegramNotif interface {
22	Send(chanID string, item feed.Item) error
23	}
24
25	// TwitterNotif is interface to send message to twitter
26	type TwitterNotif interface {
27	Send(item feed.Item) error
28	}
29
30	// Processor is a feed reader and store writer
31	type Processor struct {
32	Conf *config.Conf
33	Store *BoltDB
34	TelegramNotif TelegramNotif
35	TwitterNotif TwitterNotif
36	}
37
38	// Do activate loop of goroutine for each feed, concurrency limited by p.Conf.Concurrent
39	func (p *Processor) Do(ctx context.Context) error {	4✔
40	log.Printf("[INFO] activate processor, feeds=%d, %+v", len(p.Conf.Feeds), p.Conf)	4✔
41		4✔
42	for {	16✔
43	select {	12✔
44	case <-ctx.Done():	4✔
45	return ctx.Err()	4✔
46	default:	8✔
47	p.processFeeds(ctx)	8✔
48	}
49	}
50	}
51
52	func (p *Processor) processFeeds(ctx context.Context) {	8✔
53	log.Printf("[DEBUG] refresh started")	8✔
54	swg := syncs.NewSizedGroup(p.Conf.System.Concurrent, syncs.Preemptive, syncs.Context(ctx))	8✔
55	for name, fm := range p.Conf.Feeds {	16✔
56	for _, src := range fm.Sources {	16✔
57	name, src, fm := name, src, fm	8✔
58	swg.Go(func(context.Context) {	16✔
59	p.processFeed(name, src.URL, fm.TelegramChannel, p.Conf.System.MaxItems, fm.Filter)	8✔
60	})	8✔
61	}
62	}
63	swg.Wait()	8✔
64	log.Printf("[DEBUG] refresh completed")	8✔
65	time.Sleep(p.Conf.System.UpdateInterval)	8✔
66	}
67
68	func (p *Processor) processFeed(name, url, telegramChannel string, maximum int, filter config.Filter) {	8✔
69	rss, err := feed.Parse(url)	8✔
70	if err != nil {	8✔
71	log.Printf("[WARN] failed to parse %s, %v", url, err)	×
72	return	×
73	}	×
74
75	// up to MaxItems (5) items from each feed
76	upto := maximum	8✔
77	if len(rss.ItemList) <= maximum {	12✔
78	upto = len(rss.ItemList)	4✔
79	}	4✔
80
81	for _, item := range rss.ItemList[:upto] {	40✔
82	// skip 1y and older	32✔
83	if item.DT.Before(time.Now().AddDate(-1, 0, 0)) {	36✔
84	continue	4✔
85	}
86
87	skip, err := filter.Skip(item)	28✔
88	if err != nil {	28✔
89	log.Printf("[WARN] failed to filter %s (%s) to %s, save as is, %v", item.GUID, item.PubDate, name, err)	×
90	}	×
91	if skip {	30✔
92	item.Junk = true	2✔
93	log.Printf("[INFO] filtered %s (%s), %s %s", item.GUID, item.PubDate, name, item.Title)	2✔
94	}	2✔
95
96	created, err := p.Store.Save(name, item)	28✔
97	if err != nil {	28✔
98	log.Printf("[WARN] failed to save %s (%s) to %s, %v", item.GUID, item.PubDate, name, err)	×
99	}	×
100
101	// don't attempt to send anything if the entry was already saved
102	// or in case it was filtered out
103	if !created \|\| item.Junk {	45✔
104	continue	17✔
105	}
106
107	rptr := repeater.NewDefault(3, 5*time.Second)	11✔
108	attemptNum := 0	11✔
109	err = rptr.Do(context.Background(), func() error {	22✔
110	attemptNum++	11✔
111	startTime := time.Now()	11✔
112	log.Printf("[DEBUG] sending telegram message (attempt %d/3): title=%q, size=%d bytes, url=%s to channel=%s",	11✔
113	attemptNum, item.Title, item.Enclosure.Length, item.Enclosure.URL, telegramChannel)	11✔
114		11✔
115	if e := p.TelegramNotif.Send(telegramChannel, item); e != nil {	11✔
NEW 116	elapsed := time.Since(startTime)	×
NEW 117	log.Printf("[WARN] failed attempt %d/3 to send telegram message after %v: title=%q, size=%d bytes, url=%s to channel=%s, error=%v",	×
NEW 118	attemptNum, elapsed, item.Title, item.Enclosure.Length, item.Enclosure.URL, telegramChannel, e)	×
119	return e	×
120	}	×
121
122	elapsed := time.Since(startTime)	11✔
123	log.Printf("[INFO] successfully sent telegram message in %v: title=%q, size=%d bytes",	11✔
124	elapsed, item.Title, item.Enclosure.Length)	11✔
125	return nil	11✔
126	})
127	if err != nil {	11✔
NEW 128	log.Printf("[WARN] failed to send telegram message after 3 attempts: title=%q, size=%d bytes, url=%s to channel=%s, final_error=%v",	×
NEW 129	item.Title, item.Enclosure.Length, item.Enclosure.URL, telegramChannel, err)	×
UNCOV 130	}	×
131
132	if err := p.TwitterNotif.Send(item); err != nil {	11✔
133	log.Printf("[WARN] failed send twitter message, url=%s, %v", item.Enclosure.URL, err)	×
134	}	×
135	}
136
137	// keep up to MaxKeepInDB items in bucket
138	if removed, err := p.Store.removeOld(name, p.Conf.System.MaxKeepInDB); err == nil {	16✔
139	if removed > 0 {	9✔
140	log.Printf("[DEBUG] removed %d from %s", removed, name)	1✔
141	}	1✔
142	} else {	×
143	log.Printf("[WARN] failed to remove, %v", err)	×
144	}	×
145	}

umputun / feed-master / 16810397243

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous