Simplify concurrency in retrieving gitlab source response

4 files changed, 36 insertions, 41 deletions

diff --git a/internal/source/gitlab/cache/cache.go b/internal/source/gitlab/cache/cache.go
index b3749450..637f27e3 100644
--- a/internal/source/gitlab/cache/cache.go
+++ b/internal/source/gitlab/cache/cache.go
@@ -32,7 +32,5 @@ func (c *Cache) Resolve(ctx context.Context, domain string) *Lookup {
 		return entry.Lookup()
 	}
 
-	<-entry.Retrieve(ctx, c.client)
-
-	return entry.Lookup()
+	return entry.Retrieve(ctx, c.client)
 }
diff --git a/internal/source/gitlab/cache/cache_test.go b/internal/source/gitlab/cache/cache_test.go
index 3a5a2e8f..53e61582 100644
--- a/internal/source/gitlab/cache/cache_test.go
+++ b/internal/source/gitlab/cache/cache_test.go
@@ -67,9 +67,7 @@ func (cache *Cache) withTestEntry(config entryConfig, block func(*Entry)) {
 	entry := cache.store.LoadOrCreate(domain)
 
 	if config.retrieved {
-		newResponse := make(chan Lookup, 1)
-		newResponse <- Lookup{Domain: Domain{Name: domain}, Status: 200}
-		entry.setResponse(newResponse)
+		entry.setResponse(Lookup{Domain: Domain{Name: domain}, Status: 200})
 	}
 
 	if config.expired {
diff --git a/internal/source/gitlab/cache/entry.go b/internal/source/gitlab/cache/entry.go
index 01f44a47..67a1a03a 100644
--- a/internal/source/gitlab/cache/entry.go
+++ b/internal/source/gitlab/cache/entry.go
@@ -2,6 +2,7 @@ package cache
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"time"
 )
@@ -17,7 +18,7 @@ var (
 type Entry struct {
 	domain    string
 	created   time.Time
-	fetch     *sync.Once
+	retrieve  *sync.Once
 	refresh   *sync.Once
 	mux       *sync.RWMutex
 	retrieved chan struct{}
@@ -28,7 +29,7 @@ func newCacheEntry(domain string) *Entry {
 	return &Entry{
 		domain:    domain,
 		created:   time.Now(),
-		fetch:     &sync.Once{},
+		retrieve:  &sync.Once{},
 		refresh:   &sync.Once{},
 		mux:       &sync.RWMutex{},
 		retrieved: make(chan struct{}),
@@ -61,17 +62,22 @@ func (e *Entry) Lookup() *Lookup {
 	return e.response
 }
 
-// Retrieve schedules a retrieval of a response. It returns a channel that is
-// going to be closed when retrieval is done, either successfully or not.
-func (e *Entry) Retrieve(ctx context.Context, client Resolver) <-chan struct{} {
-	// TODO create context with timeout
-	e.fetch.Do(func() {
-		retriever := Retriever{client: client, ctx: ctx, timeout: retrievalTimeout}
+// Retrieve perform a blocking retrieval of the cache entry response.
+func (e *Entry) Retrieve(ctx context.Context, client Resolver) *Lookup {
+	newctx, cancelctx := context.WithTimeout(ctx, retrievalTimeout)
+	defer cancelctx()
 
-		go e.setResponse(retriever.Retrieve(e.domain))
-	})
+	e.retrieve.Do(func() { go e.retrieveWithClient(client) })
+
+	select {
+	case <-newctx.Done():
+		return &Lookup{Status: 502, Error: errors.New("context done")}
+	case <-e.retrieved:
+		return e.Lookup()
+	}
 
-	return e.retrieved
+	// This should not happen
+	return &Lookup{Status: 500, Error: errors.New("retrieval error")}
 }
 
 // Refresh will update the entry in the store only when it gets resolved.
@@ -80,21 +86,24 @@ func (e *Entry) Refresh(client Resolver, store Store) {
 		go func() {
 			entry := newCacheEntry(e.domain)
 
-			<-entry.Retrieve(context.Background(), client)
+			entry.Retrieve(context.Background(), client)
 
 			store.ReplaceOrCreate(e.domain, entry)
 		}()
 	})
 }
 
-func (e *Entry) setResponse(response <-chan Lookup) {
-	lookup := <-response
+func (e *Entry) retrieveWithClient(client Resolver) {
+	retriever := Retriever{client: client, timeout: retrievalTimeout}
 
+	e.setResponse(retriever.Retrieve(e.domain))
+}
+
+func (e *Entry) setResponse(lookup Lookup) {
 	e.mux.Lock()
 	defer e.mux.Unlock()
 
 	e.response = &lookup
-
 	close(e.retrieved)
 }
 
diff --git a/internal/source/gitlab/cache/retriever.go b/internal/source/gitlab/cache/retriever.go
index de766aa4..e57565ae 100644
--- a/internal/source/gitlab/cache/retriever.go
+++ b/internal/source/gitlab/cache/retriever.go
@@ -13,42 +13,32 @@ var maxRetrievalInterval = time.Second
 // case of errors
 type Retriever struct {
 	client  Resolver
-	ctx     context.Context
 	timeout time.Duration
 }
 
-// Retrieve schedules a retrieval of a response and return a channel that the
-// response is going to be sent to
-func (r *Retriever) Retrieve(domain string) <-chan Lookup {
-	response := make(chan Lookup)
-
-	go r.retrieveWithTimeout(domain, response)
-
-	return response
-}
-
-func (r *Retriever) retrieveWithTimeout(domain string, response chan<- Lookup) {
-	newctx, cancel := context.WithTimeout(r.ctx, r.timeout)
+// Retrieve retrieves a lookup response from external source with timeout and
+// backoff. It has its own context with timeout.
+func (r *Retriever) Retrieve(domain string) Lookup {
+	ctx, cancel := context.WithTimeout(context.Background(), r.timeout)
 	defer cancel()
 
 	var lookup Lookup
 
 	select {
-	case <-newctx.Done():
-		response <- Lookup{Status: 502, Error: errors.New("context done")}
+	case <-ctx.Done():
 		fmt.Println("retrieval context done") // TODO logme
-	case lookup = <-r.resolveWithBackoff(newctx, domain):
-		response <- lookup
+		lookup = Lookup{Status: 502, Error: errors.New("retrieval context done")}
+	case lookup = <-r.resolveWithBackoff(ctx, domain):
 		fmt.Println("retrieval response sent") // TODO logme
 	}
 
-	close(response)
+	return lookup
 }
 
 func (r *Retriever) resolveWithBackoff(ctx context.Context, domain string) <-chan Lookup {
 	response := make(chan Lookup)
 
-	go func(response chan<- Lookup) {
+	go func() {
 		var lookup Lookup
 
 		for i := 1; i <= 3; i++ {
@@ -63,7 +53,7 @@ func (r *Retriever) resolveWithBackoff(ctx context.Context, domain string) <-cha
 
 		response <- lookup
 		close(response)
-	}(response)
+	}()
 
 	return response
 }