fix sleep on non-renewable vault secrets

An over-zealous refactoring resulted in <-time.After() channel used for
non-renewable sleep to be ignored. I originally had channel of the time
channels but "simplified" it too much. This reworks it the way the sleep
is passed to the next iteration to just pass the time.Duration of the
sleep down the channel, then time.Sleep-ing when reading off the
channel. This way it can see it has a sleep to do, then do it instead of
skipping it as the time hadn't passed yet.

This also let me write a test for it as I don't need to actually sleep
to see that there is something in the channel.

Fixes #1272

dependency/vault_read.go

@@ -19,7 +19,7 @@ var (
 // VaultReadQuery is the dependency to Vault for a secret
 type VaultReadQuery struct {
 	stopCh  chan struct{}
-	sleepCh <-chan time.Time
+	sleepCh chan time.Duration
 
 	rawPath     string
 	queryValues url.Values
@@ -46,7 +46,7 @@ func NewVaultReadQuery(s string) (*VaultReadQuery, error) {
 
 	return &VaultReadQuery{
 		stopCh:      make(chan struct{}, 1),
-		sleepCh:     make(chan time.Time, 1),
+		sleepCh:     make(chan time.Duration, 1),
 		rawPath:     secretURL.Path,
 		queryValues: secretURL.Query(),
 	}, nil
@@ -61,7 +61,8 @@ func (d *VaultReadQuery) Fetch(clients *ClientSet, opts *QueryOptions,
 	default:
 	}
 	select {
-	case <-d.sleepCh:
+	case dur := <-d.sleepCh:
+		time.Sleep(dur)
 	default:
 	}
 
@@ -82,7 +83,7 @@ func (d *VaultReadQuery) Fetch(clients *ClientSet, opts *QueryOptions,
 	if !vaultSecretRenewable(d.secret) {
 		dur := leaseCheckWait(d.secret)
 		log.Printf("[TRACE] %s: non-renewable secret, set sleep for %s", d, dur)
-		d.sleepCh = time.After(dur)
+		d.sleepCh <- dur
 	}
 
 	return respWithMetadata(d.secret)

dependency/vault_read_test.go

@@ -227,6 +227,39 @@ func TestVaultReadQuery_Fetch_KVv1(t *testing.T) {
 		case <-dataCh:
 		}
 	})
+
+	t.Run("nonrenewable-sleeper", func(t *testing.T) {
+		d, err := NewVaultReadQuery(secretsPath + "/foo/bar")
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		_, qm, err := d.Fetch(clients, nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		errCh := make(chan error, 1)
+		go func() {
+			_, _, err := d.Fetch(clients,
+				&QueryOptions{WaitIndex: qm.LastIndex})
+			if err != nil {
+				errCh <- err
+			}
+			close(errCh)
+		}()
+
+		if err := <-errCh; err != nil {
+			t.Fatal(err)
+		}
+		if len(d.sleepCh) != 1 {
+			t.Fatalf("sleep channel has len %v, expected 1", len(d.sleepCh))
+		}
+		dur := <-d.sleepCh
+		if dur > 0 {
+			t.Fatalf("duration of sleep should be > 0")
+		}
+	})
 }
 
 func TestVaultReadQuery_Fetch_KVv2(t *testing.T) {

dependency/vault_write.go

@@ -22,7 +22,7 @@ var (
 // VaultWriteQuery is the dependency to Vault for a secret
 type VaultWriteQuery struct {
 	stopCh  chan struct{}
-	sleepCh <-chan time.Time
+	sleepCh chan time.Duration
 
 	path     string
 	data     map[string]interface{}
@@ -43,7 +43,7 @@ func NewVaultWriteQuery(s string, d map[string]interface{}) (*VaultWriteQuery, e
 
 	return &VaultWriteQuery{
 		stopCh:   make(chan struct{}, 1),
-		sleepCh:  make(chan time.Time, 1),
+		sleepCh:  make(chan time.Duration, 1),
 		path:     s,
 		data:     d,
 		dataHash: sha1Map(d),
@@ -59,7 +59,8 @@ func (d *VaultWriteQuery) Fetch(clients *ClientSet, opts *QueryOptions,
 	default:
 	}
 	select {
-	case <-d.sleepCh:
+	case dur := <-d.sleepCh:
+		time.Sleep(dur)
 	default:
 	}
 
@@ -91,7 +92,7 @@ func (d *VaultWriteQuery) Fetch(clients *ClientSet, opts *QueryOptions,
 	if !vaultSecretRenewable(d.secret) {
 		dur := leaseCheckWait(d.secret)
 		log.Printf("[TRACE] %s: non-renewable secret, set sleep for %s", d, dur)
-		d.sleepCh = time.After(dur)
+		d.sleepCh <- dur
 	}
 
 	return respWithMetadata(d.secret)

dependency/vault_write_test.go

@@ -292,6 +292,42 @@ func TestVaultWriteQuery_Fetch(t *testing.T) {
 		case <-dataCh:
 		}
 	})
+
+	t.Run("nonrenewable-sleeper", func(t *testing.T) {
+		d, err := NewVaultWriteQuery("transit/encrypt/test",
+			map[string]interface{}{
+				"plaintext": b64("test"),
+			})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		_, qm, err := d.Fetch(clients, nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		errCh := make(chan error, 1)
+		go func() {
+			_, _, err := d.Fetch(clients,
+				&QueryOptions{WaitIndex: qm.LastIndex})
+			if err != nil {
+				errCh <- err
+			}
+			close(errCh)
+		}()
+
+		if err := <-errCh; err != nil {
+			t.Fatal(err)
+		}
+		if len(d.sleepCh) != 1 {
+			t.Fatalf("sleep channel has len %v, expected 1", len(d.sleepCh))
+		}
+		dur := <-d.sleepCh
+		if dur > 0 {
+			t.Fatalf("duration of sleep should be > 0")
+		}
+	})
 }
 
 func TestVaultWriteQuery_String(t *testing.T) {