Merge pull request #11 from getlantern/afisk/initfrom-file-modtime

Record a cached file's age, not its load time, when initializing from it

Author: myleshorton

initfrom_test.go

@@ -0,0 +1,185 @@
+package keepcurrent
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// lastModifiedServer serves body with Last-Modified mod, answering 304 to an
+// If-Modified-Since at or after it, as a CDN or object store does.
+func lastModifiedServer(t *testing.T, mod time.Time, body string) *httptest.Server {
+	t.Helper()
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if ims, err := http.ParseTime(r.Header.Get("If-Modified-Since")); err == nil && !mod.After(ims) {
+			w.WriteHeader(http.StatusNotModified)
+			return
+		}
+		w.Header().Set("Last-Modified", mod.UTC().Format(http.TimeFormat))
+		_, _ = w.Write([]byte(body))
+	}))
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+// A runner started from a cached file older than the web copy fetches the web
+// copy, even when the process starts after the web copy was published. Before,
+// InitFrom recorded the start time as the data's age, so the first web check
+// asked only for data newer than the start, got 304, and kept the old file
+// until the web copy changed again.
+func TestInitFromOlderFileFetchesNewerWebCopy(t *testing.T) {
+	published := time.Now().Add(-time.Hour).Truncate(time.Second)
+	srv := lastModifiedServer(t, published, "new")
+
+	path := filepath.Join(t.TempDir(), "db")
+	require.NoError(t, os.WriteFile(path, []byte("old"), 0644))
+	cached := published.Add(-24 * time.Hour)
+	require.NoError(t, os.Chtimes(path, cached, cached))
+
+	ch := make(chan []byte, 4)
+	runner := New(FromWeb(srv.URL), ToFile(path), ToChannel(ch))
+	runner.InitFrom(FromFile(path))
+	assert.Equal(t, "old", string(<-ch), "InitFrom loads the cached file")
+	stop := runner.Start(time.Hour)
+	defer stop()
+
+	select {
+	case got := <-ch:
+		assert.Equal(t, "new", string(got), "the first web check fetches the newer copy")
+	case <-time.After(5 * time.Second):
+		t.Fatal("the runner never fetched the web copy published after its cached file")
+	}
+	b, err := os.ReadFile(path)
+	require.NoError(t, err)
+	assert.Equal(t, "new", string(b), "and saves it")
+}
+
+// A runner started from a cached file at least as new as the web copy doesn't
+// fetch it again.
+func TestInitFromCurrentFileSkipsTheWebCopy(t *testing.T) {
+	published := time.Now().Add(-time.Hour).Truncate(time.Second)
+	srv := lastModifiedServer(t, published, "new")
+
+	path := filepath.Join(t.TempDir(), "db")
+	require.NoError(t, os.WriteFile(path, []byte("cached"), 0644))
+	cached := published.Add(time.Minute)
+	require.NoError(t, os.Chtimes(path, cached, cached))
+
+	ch := make(chan []byte, 4)
+	runner := New(FromWeb(srv.URL), ToFile(path), ToChannel(ch))
+	runner.InitFrom(FromFile(path))
+	assert.Equal(t, "cached", string(<-ch))
+	stop := runner.Start(time.Hour)
+	time.Sleep(200 * time.Millisecond)
+	stop()
+	select {
+	case got := <-ch:
+		t.Fatalf("fetched %q although the cached file is current", got)
+	default:
+	}
+}
+
+// InitFrom doesn't write the file it read back to itself, so the file keeps
+// the age that tells the next runner whether it is current.
+func TestInitFromKeepsTheCachedFilesAge(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "db")
+	require.NoError(t, os.WriteFile(path, []byte("cached"), 0644))
+	cached := time.Now().Add(-48 * time.Hour).Truncate(time.Second)
+	require.NoError(t, os.Chtimes(path, cached, cached))
+
+	ch := make(chan []byte, 1)
+	runner := New(FromWeb("http://127.0.0.1:1/unused"), ToFile(path), ToChannel(ch))
+	runner.InitFrom(FromFile(path))
+	assert.Equal(t, "cached", string(<-ch))
+
+	fi, err := os.Stat(path)
+	require.NoError(t, err)
+	assert.True(t, fi.ModTime().Equal(cached), "modtime %v, want %v", fi.ModTime(), cached)
+}
+
+// If-Modified-Since is the cutoff in GMT, whatever the host's time zone.
+func TestWebSourceSendsIfModifiedSinceInGMT(t *testing.T) {
+	var got string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		got = r.Header.Get("If-Modified-Since")
+		w.WriteHeader(http.StatusNotModified)
+	}))
+	t.Cleanup(srv.Close)
+	cutoff := time.Date(2026, 10, 8, 17, 15, 0, 0, time.FixedZone("UTC-6", -6*3600))
+	_, err := FromWeb(srv.URL).Fetch(cutoff)
+	assert.ErrorIs(t, err, ErrUnmodified)
+	assert.Equal(t, "Thu, 08 Oct 2026 23:15:00 GMT", got)
+}
+
+// A sink that preprocesses still writes to the file the source read, since
+// its output can differ from what was read.
+func TestInitFromStillWritesAPreprocessedCopyInPlace(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "db")
+	require.NoError(t, os.WriteFile(path, []byte("cached"), 0644))
+	upper := func(r io.Reader) (io.Reader, error) {
+		b, err := io.ReadAll(r)
+		return strings.NewReader(strings.ToUpper(string(b))), err
+	}
+	runner := New(FromWeb("http://127.0.0.1:1/unused"), ToFileWithPreprocessor(path, upper))
+	runner.InitFrom(FromFile(path))
+	b, err := os.ReadFile(path)
+	require.NoError(t, err)
+	assert.Equal(t, "CACHED", string(b))
+}
+
+// A cached file dated in the future says nothing about its age, so the first
+// web check is unconditional and fetches a copy published before that date.
+func TestInitFromFutureDatedFileFetchesTheWebCopy(t *testing.T) {
+	published := time.Now().Add(-time.Hour).Truncate(time.Second)
+	srv := lastModifiedServer(t, published, "new")
+
+	path := filepath.Join(t.TempDir(), "db")
+	require.NoError(t, os.WriteFile(path, []byte("old"), 0644))
+	future := time.Now().Add(48 * time.Hour)
+	require.NoError(t, os.Chtimes(path, future, future))
+
+	ch := make(chan []byte, 4)
+	runner := New(FromWeb(srv.URL), ToFile(path), ToChannel(ch))
+	runner.InitFrom(FromFile(path))
+	assert.Equal(t, "old", string(<-ch))
+	stop := runner.Start(time.Hour)
+	defer stop()
+	select {
+	case got := <-ch:
+		assert.Equal(t, "new", string(got))
+	case <-time.After(5 * time.Second):
+		t.Fatal("a future-dated cached file kept the runner from fetching the web copy")
+	}
+}
+
+// A sink that reaches the source's file through a symlink is the same file, so
+// InitFrom doesn't rewrite it either.
+func TestInitFromKeepsTheAgeThroughASymlink(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "db")
+	link := filepath.Join(dir, "db-link")
+	require.NoError(t, os.WriteFile(path, []byte("cached"), 0644))
+	require.NoError(t, os.Symlink(path, link))
+	cached := time.Now().Add(-48 * time.Hour).Truncate(time.Second)
+	require.NoError(t, os.Chtimes(path, cached, cached))
+
+	ch := make(chan []byte, 1)
+	runner := New(FromWeb("http://127.0.0.1:1/unused"), ToFile(link), ToChannel(ch))
+	runner.InitFrom(FromFile(path))
+	assert.Equal(t, "cached", string(<-ch))
+
+	fi, err := os.Lstat(link)
+	require.NoError(t, err)
+	assert.NotZero(t, fi.Mode()&os.ModeSymlink, "the link is still a link")
+	fi, err = os.Stat(path)
+	require.NoError(t, err)
+	assert.True(t, fi.ModTime().Equal(cached), "modtime %v, want %v", fi.ModTime(), cached)
+}

keepcurrent.go

@@ -101,18 +101,28 @@ func (runner *Runner) syncOnce(from Source, chStop chan struct{}) {
 		if err == ErrUnmodified {
 			return
 		}
+		updated := start
 		if err == nil {
 			// Read ahead to surface any error reading from the source. readAll
 			// pre-sizes its buffer when the source reports a length, avoiding the
 			// reallocation churn io.ReadAll incurs on large payloads.
 			data, err = readAll(rc)
 			rc.Close()
+			// Data read from a file is as old as the file, not as the read. A
+			// modification time after the read (clock skew, a copied file) says
+			// nothing about the data's age, so the next check is unconditional.
+			if m, ok := rc.(interface{ sourceModTime() time.Time }); ok {
+				updated = m.sourceModTime()
+				if updated.After(start) {
+					updated = time.Time{}
+				}
+			}
 		}
 		if err == nil {
 			err = runner.Validate(data)
 		}
 		if err == nil {
-			runner.lastUpdated = start
+			runner.lastUpdated = updated
 			break
 		}
 		d := runner.OnSourceError(err, tries)
@@ -126,6 +136,9 @@ func (runner *Runner) syncOnce(from Source, chStop chan struct{}) {
 		}
 	}
 	for _, s := range runner.sinks {
+		if writesBack(from, s) {
+			continue
+		}
 		if err := s.UpdateFrom(bytes.NewReader(data)); err != nil {
 			runner.OnSinkError(s, err)
 		}

source.go

@@ -8,6 +8,7 @@ import (
 	"net/http"
 	"os"
 	"path"
+	"path/filepath"
 	"sync"
 	"time"
 
@@ -46,7 +47,9 @@ func (s *webSource) Fetch(ifNewerThan time.Time) (io.ReadCloser, error) {
 		return nil, err
 	}
 	if !ifNewerThan.IsZero() {
-		req.Header.Add("If-Modified-Since", ifNewerThan.Format(http.TimeFormat))
+		// http.TimeFormat is GMT, so the time must be in UTC first; formatted in
+		// local time it would claim a different instant on any host not on UTC.
+		req.Header.Add("If-Modified-Since", ifNewerThan.UTC().Format(http.TimeFormat))
 	}
 	if s.getETag() != "" {
 		req.Header.Add("If-None-Match", s.etag)
@@ -211,11 +214,58 @@ func (s *fileSource) Fetch(ifNewerThan time.Time) (io.ReadCloser, error) {
 		// The preprocessor may hand back a stream that doesn't close the file it
 		// wraps (e.g. io.NopCloser), so couple Close to the underlying file to
 		// avoid leaking the descriptor.
-		return fileBackedReadCloser{ReadCloser: result, f: f}, nil
+		return modTimeReadCloser{ReadCloser: fileBackedReadCloser{ReadCloser: result, f: f}, mod: fi.ModTime()}, nil
 	}
 	// Surface the file size so the Runner's readAll pre-sizes its buffer instead
 	// of growing by reallocation. The cached payloads read here are exactly the
 	// large ones that churn — e.g. geo loads a ~75MB MaxMind mmdb from disk via
 	// InitFrom(FromFile(...)) at startup.
-	return sizedReadCloser{ReadCloser: f, n: fi.Size()}, nil
+	return modTimeReadCloser{ReadCloser: sizedReadCloser{ReadCloser: f, n: fi.Size()}, mod: fi.ModTime()}, nil
+}
+
+// modTimeReadCloser is a file's contents with the file's modification time, so
+// a Runner records how old the data it loaded is rather than when it loaded it.
+// Without this, a Runner initialised from an older cached file would ask its
+// web source only for data newer than the moment it started, and never fetch
+// an update published between the file's download and that moment.
+type modTimeReadCloser struct {
+	io.ReadCloser
+	mod time.Time
+}
+
+func (r modTimeReadCloser) sourceModTime() time.Time { return r.mod }
+
+// size passes through the wrapped reader's size, or -1 (unknown) when it has
+// none, so readAll still pre-sizes a plain file read.
+func (r modTimeReadCloser) size() int64 {
+	if n, ok := knownSize(r.ReadCloser); ok {
+		return n
+	}
+	return -1
+}
+
+// writesBack reports whether sink s would copy source from's file onto itself
+// unchanged, which a sync skips: the rewrite changes nothing but the file's
+// modification time, which moves to now, so the next Runner to start from it
+// would take an old file for a fresh one. A preprocessor on either side can
+// change the contents, so those writes still happen.
+func writesBack(from Source, s Sink) bool {
+	src, ok := from.(*fileSource)
+	if !ok || src.preprocessor != nil {
+		return false
+	}
+	dst, ok := s.(*fileSink)
+	if !ok || dst.preprocessor != nil {
+		return false
+	}
+	// Two paths can name one file (a symlink, a relative path): compare the
+	// files themselves when both exist.
+	if srcInfo, err := os.Stat(src.path); err == nil {
+		if dstInfo, err := os.Stat(dst.path); err == nil {
+			return os.SameFile(srcInfo, dstInfo)
+		}
+	}
+	a, errA := filepath.Abs(src.path)
+	b, errB := filepath.Abs(dst.path)
+	return errA == nil && errB == nil && a == b
 }