diff --git a/gtfsdb/conditional_import_test.go b/gtfsdb/conditional_import_test.go index 4406a5087..ec995e9fd 100644 --- a/gtfsdb/conditional_import_test.go +++ b/gtfsdb/conditional_import_test.go @@ -325,3 +325,33 @@ func TestClearAllGTFSData(t *testing.T) { require.NoError(t, err, "Import metadata should still exist after clear") assert.NotEmpty(t, metadata.FileHash, "Import metadata should not be cleared") } + +func TestSkipIfImported(t *testing.T) { + client, err := NewClient(Config{DBPath: ":memory:", Env: appconf.Test}) + require.NoError(t, err) + t.Cleanup(func() { _ = client.Close() }) + + imported, err := client.SkipIfImported(t.Context(), "hash", "source") + require.NoError(t, err) + assert.False(t, imported, "an empty database holds no feed") + + raba, err := os.ReadFile("../testdata/raba.zip") + require.NoError(t, err) + parsed, err := ParseGtfsData(raba, "source") + require.NoError(t, err) + _, err = client.StoreGtfsData(t.Context(), parsed) + require.NoError(t, err) + + for _, tt := range []struct { + hash, source string + want bool + }{ + {parsed.Hash, "source", true}, + {parsed.Hash, "other-source", false}, + {"other-hash", "source", false}, + } { + imported, err := client.SkipIfImported(t.Context(), tt.hash, tt.source) + require.NoError(t, err) + assert.Equal(t, tt.want, imported, "%s from %s", tt.hash, tt.source) + } +} diff --git a/gtfsdb/helpers.go b/gtfsdb/helpers.go index a8d068d5a..3c1c059bb 100644 --- a/gtfsdb/helpers.go +++ b/gtfsdb/helpers.go @@ -7,6 +7,7 @@ import ( "database/sql" _ "embed" "encoding/hex" + "errors" "fmt" "iter" "log/slog" @@ -23,9 +24,9 @@ import ( //go:embed schema.sql var ddl string -// GtfsData bundles parsed static GTFS data with the sha256 hash of the -// bytes it was parsed from. The hash is used to skip reimport when the -// source data is unchanged. +// GtfsData bundles parsed static GTFS data with the hash that identifies the +// feed in import_metadata. The hash is used to skip reimport when the feed is +// unchanged. type GtfsData struct { Static *gtfs.Static Hash string @@ -36,8 +37,12 @@ type GtfsData struct { // The given source is stored on the returned struct for later use by StoreGtfsData. func ParseGtfsData(b []byte, source string) (*GtfsData, error) { hash := sha256.Sum256(b) - hashStr := hex.EncodeToString(hash[:]) + return ParseGtfsDataWithHash(b, hex.EncodeToString(hash[:]), source) +} +// ParseGtfsDataWithHash parses and structurally validates GTFS zip bytes whose +// identifying hash the caller has already computed. +func ParseGtfsDataWithHash(b []byte, hash, source string) (*GtfsData, error) { logger := slog.Default().With(slog.String("component", "gtfs_parser")) logging.LogOperation(logger, "parsing_gtfs_data", slog.Int("bytes", len(b))) staticData, err := gtfs.ParseStatic(b, gtfs.ParseStaticOptions{}) @@ -49,7 +54,7 @@ func ParseGtfsData(b []byte, source string) (*GtfsData, error) { return nil, fmt.Errorf("GTFS validation failed: %w", err) } - return &GtfsData{Static: staticData, Hash: hashStr, Source: source}, nil + return &GtfsData{Static: staticData, Hash: hash, Source: source}, nil } // metricsWrapper wraps *sql.DB for metric reporting purposes @@ -203,6 +208,49 @@ func (c *Client) withTransaction(ctx context.Context, tx *sql.Tx, label string, return nil } +// SkipIfImported reports whether the database already holds the feed with +// this hash from this source, so a caller can skip parsing and storing it. +// When it does, it refreshes the feed's derived metadata, as StoreGtfsData +// does for an unchanged feed. +func (c *Client) SkipIfImported(ctx context.Context, hash, source string) (bool, error) { + metadata, err := c.Queries.GetImportMetadata(ctx) + if errors.Is(err, sql.ErrNoRows) { + return false, nil + } + if err != nil { + return false, fmt.Errorf("error checking import metadata: %w", err) + } + return c.skipUnchangedFeed(ctx, metadata, hash, source) +} + +// skipUnchangedFeed is the one rule for a feed the database already holds. +// When import_metadata describes the feed with this hash from this source, it +// refreshes the derived metadata and reports true; otherwise it does nothing. +func (c *Client) skipUnchangedFeed(ctx context.Context, metadata ImportMetadatum, hash, source string) (bool, error) { + if metadata.FileHash != hash || metadata.FileSource != source { + return false, nil + } + if err := c.refreshFeedExpiry(ctx); err != nil { + return false, err + } + logger := slog.Default().With(slog.String("component", "gtfs_importer")) + logging.LogOperation(logger, "gtfs_data_unchanged_skipping_import", slog.String("hash", hash[:8])) + return true, nil +} + +// refreshFeedExpiry recomputes feed_expires_at from the stored calendar. It +// runs when a feed is found unchanged, so existing databases pick up +// corrections to the expiry calculation on upgrade without a re-import. +func (c *Client) refreshFeedExpiry(ctx context.Context) error { + err := c.withTransaction(ctx, nil, "refresh_feed_expiry", func(tx *sql.Tx) error { + return updateFeedExpiresAtFromCalendar(ctx, c.Queries.WithTx(tx)) + }) + if err != nil { + return fmt.Errorf("failed to refresh feed expiry: %w", err) + } + return nil +} + func (c *Client) StoreGtfsData(ctx context.Context, data *GtfsData) (bool, error) { logger := slog.Default().With(slog.String("component", "gtfs_importer")) @@ -222,17 +270,11 @@ func (c *Client) StoreGtfsData(ctx context.Context, data *GtfsData) (bool, error existingMetadata, err := c.Queries.GetImportMetadata(ctx) if err == nil { hasExisting = true - // We have existing metadata, check if hash matches - if existingMetadata.FileHash == data.Hash && existingMetadata.FileSource == data.Source { - // Refresh derived metadata even when the feed is unchanged, so existing - // databases pick up corrections to the expiry calculation on upgrade. - if err := c.withTransaction(ctx, nil, "refresh_feed_expiry", func(tx *sql.Tx) error { - return updateFeedExpiresAtFromCalendar(ctx, c.Queries.WithTx(tx)) - }); err != nil { - return false, fmt.Errorf("failed to refresh feed expiry: %w", err) - } - logging.LogOperation(logger, "gtfs_data_unchanged_skipping_import", - slog.String("hash", data.Hash[:8])) + skipped, err := c.skipUnchangedFeed(ctx, existingMetadata, data.Hash, data.Source) + if err != nil { + return false, err + } + if skipped { return false, nil } } else if err != sql.ErrNoRows { diff --git a/internal/gtfs/feed_file.go b/internal/gtfs/feed_file.go new file mode 100644 index 000000000..a81e99721 --- /dev/null +++ b/internal/gtfs/feed_file.go @@ -0,0 +1,285 @@ +package gtfs + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "net/url" + "os" + "path/filepath" + "strconv" + "strings" + "time" + + "maglev.onebusaway.org/gtfsdb" + "maglev.onebusaway.org/internal/logging" +) + +// maxStaticFeedBytes is the largest static feed Maglev downloads. +const maxStaticFeedBytes = 200 * 1024 * 1024 + +// feedFile is a static GTFS zip on local disk. hash identifies it in +// import_metadata when it is imported as configured. +type feedFile struct { + path string + hash string + // private is set for a download, which nothing but this import can + // change, so hash still describes it when it is parsed. + private bool + cleanup func() +} + +// fetchFeedFile puts the configured feed in a local file and hashes it, +// without holding the feed in memory. A local feed is read where it is; a +// remote feed is downloaded next to the database. Call cleanup when done. +func fetchFeedFile(ctx context.Context, config Config) (*feedFile, error) { + logger := slog.Default().With(slog.String("component", "gtfs_loader")) + logging.LogOperation(logger, "reading_gtfs_data", slog.String("source", config.GtfsURL)) + + if config.isLocalFile() { + contentHash, err := hashFile(config.GtfsURL) + if err != nil { + return nil, fmt.Errorf("error reading local GTFS file: %w", err) + } + return &feedFile{path: config.GtfsURL, hash: feedHash(contentHash, config), cleanup: func() {}}, nil + } + return downloadFeedFile(ctx, config) +} + +func hashFile(path string) (string, error) { + file, err := os.Open(path) + if err != nil { + return "", err + } + defer func() { _ = file.Close() }() + + hasher := sha256.New() + if _, err := io.Copy(hasher, file); err != nil { + return "", err + } + return hex.EncodeToString(hasher.Sum(nil)), nil +} + +func downloadFeedFile(ctx context.Context, config Config) (*feedFile, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, config.GtfsURL, nil) + if err != nil { + return nil, fmt.Errorf("error creating GTFS request: %w", err) + } + if config.StaticAuthHeaderKey != "" && config.StaticAuthHeaderValue != "" { + req.Header.Set(config.StaticAuthHeaderKey, config.StaticAuthHeaderValue) + } + + resp, err := staticHTTPClient.Do(req) + if err != nil { + return nil, fmt.Errorf("error downloading GTFS data: %w", err) + } + defer logging.SafeCloseWithLogging(resp.Body, + slog.Default().With(slog.String("component", "gtfs_downloader")), + "http_response_body") + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("failed to download GTFS data: received HTTP status %s", resp.Status) + } + + file, err := newDownloadLocation(config).create() + if err != nil { + return nil, fmt.Errorf("error creating file for GTFS download: %w", err) + } + remove := func() { _ = os.Remove(file.Name()) } + + hasher := sha256.New() + written, copyErr := io.Copy(io.MultiWriter(file, hasher), io.LimitReader(resp.Body, maxStaticFeedBytes+1)) + if err := errors.Join(copyErr, file.Close()); err != nil { + remove() + return nil, fmt.Errorf("error reading GTFS data: %w", err) + } + if written > maxStaticFeedBytes { + remove() + return nil, fmt.Errorf("static GTFS response exceeds size limit of %d bytes", maxStaticFeedBytes) + } + return &feedFile{ + path: file.Name(), + hash: feedHash(hex.EncodeToString(hasher.Sum(nil)), config), + private: true, + cleanup: remove, + }, nil +} + +// Timeouts for downloading a static feed. The overall timeout allows for a +// feed of maxStaticFeedBytes on a slow link. +const ( + staticTLSHandshakeTimeout = 10 * time.Second + staticResponseHeaderTimeout = 30 * time.Second + staticIdleConnTimeout = 90 * time.Second + staticDownloadTimeout = 5 * time.Minute +) + +// staticHTTPClient downloads static feeds. It is shared by every reload, so +// an idle connection is reused or closed by its transport rather than left +// behind on one that nothing uses again. +var staticHTTPClient = newStaticHTTPClient() + +func newStaticHTTPClient() *http.Client { + transport := cloneDefaultTransport() + transport.TLSHandshakeTimeout = staticTLSHandshakeTimeout + transport.ResponseHeaderTimeout = staticResponseHeaderTimeout + transport.IdleConnTimeout = staticIdleConnTimeout + return &http.Client{Timeout: staticDownloadTimeout, Transport: transport} +} + +// cloneDefaultTransport returns a copy of http.DefaultTransport, to keep its +// defaults (proxy from environment, dialer, HTTP/2, keepalives). +func cloneDefaultTransport() *http.Transport { + if t, ok := http.DefaultTransport.(*http.Transport); ok { + return t.Clone() + } + return &http.Transport{} +} + +// Feed downloads are named downloadPrefix, then their owner, then "-" and the +// random digits os.CreateTemp adds, then downloadSuffix. The owner part lets a +// process find the downloads it left behind in a crash without touching +// anyone else's. +const ( + downloadPrefix = "gtfs-static-" + downloadSuffix = ".zip" +) + +// sqliteMemoryDatabase is the name SQLite gives an in-memory database. +const sqliteMemoryDatabase = ":memory:" + +// downloadLocation is where a process keeps its feed downloads, and the name +// prefix that marks them as its own. +type downloadLocation struct { + dir string + prefix string +} + +// newDownloadLocation keeps downloads on the same disk as the database, rather +// than in a temporary directory that may be held in memory (tmpfs), and names +// them after the database file, so processes whose databases share a +// directory leave each other's downloads alone. With an in-memory database, +// downloads go to the temporary directory and are named after the process. +func newDownloadLocation(config Config) downloadLocation { + dbFile, onDisk := databaseFile(config.GTFSDataPath) + if !onDisk { + owner := "pid" + strconv.Itoa(os.Getpid()) + return downloadLocation{dir: os.TempDir(), prefix: downloadPrefix + owner + "-"} + } + return downloadLocation{dir: filepath.Dir(dbFile), prefix: downloadPrefix + filepath.Base(dbFile) + "-"} +} + +// databaseFile returns the file that holds the database at dataPath, a path +// or a SQLite "file:" URI, and reports whether the database is on disk at +// all rather than in memory. +func databaseFile(dataPath string) (string, bool) { + if !strings.HasPrefix(dataPath, "file:") { + inMemory := dataPath == "" || strings.HasPrefix(dataPath, sqliteMemoryDatabase) + return dataPath, !inMemory + } + uri, err := url.Parse(dataPath) + if err != nil { + return "", false + } + path := uri.Path + if path == "" { + path = uri.Opaque + } + inMemory := path == "" || path == sqliteMemoryDatabase || uri.Query().Get("mode") == "memory" + return path, !inMemory +} + +// create removes the downloads this owner left behind, then creates a new one. +func (l downloadLocation) create() (*os.File, error) { + l.removeStale() + return os.CreateTemp(l.dir, l.prefix+"*"+downloadSuffix) +} + +// removeStale deletes downloads this owner left behind in a crash. An owner's +// reloads are serialized, so none of them is still in use. +func (l downloadLocation) removeStale() { + entries, _ := os.ReadDir(l.dir) + for _, entry := range entries { + if entry.Type().IsRegular() && l.owns(entry.Name()) { + _ = os.Remove(filepath.Join(l.dir, entry.Name())) + } + } +} + +// owns reports whether name is one of this owner's downloads. The random part +// must be all digits, so another owner whose name merely starts with this one +// (gtfs.db-old next to gtfs.db) doesn't match. +func (l downloadLocation) owns(name string) bool { + random, found := strings.CutPrefix(name, l.prefix) + if !found { + return false + } + random, found = strings.CutSuffix(random, downloadSuffix) + return found && random != "" && strings.Trim(random, "0123456789") == "" +} + +// feedHash identifies a feed in import_metadata. With gtfstidy on, the stored +// data comes from the tidied feed, so the setting is part of the identity: +// turning it on or off causes a re-import. The mark records that tidy was on, +// not that it succeeded: a feed whose tidy fails is stored under the same +// hash, so the next reload skips it instead of parsing it again. Tidy is +// retried when the feed changes. +func feedHash(contentHash string, config Config) string { + if config.EnableGTFSTidy { + return contentHash + "+gtfstidy" + } + return contentHash +} + +// parse reads the feed, runs gtfstidy when enabled, then parses and validates +// it. +func (f *feedFile) parse(ctx context.Context, config Config) (*gtfsdb.GtfsData, error) { + b, hash, err := f.read(ctx, config) + if err != nil { + return nil, err + } + + data, err := gtfsdb.ParseGtfsDataWithHash(b, hash, config.GtfsURL) + if err != nil { + return nil, err + } + + if err := validateStaticAgencyTimezones(data.Static); err != nil { + return nil, fmt.Errorf("invalid GTFS agency timezone: %w", err) + } + return data, nil +} + +// read returns the feed's bytes to import, tidied when gtfstidy is on and +// succeeds, and the hash to store them under. gtfstidy reads the file itself, +// so a tidied feed is never held in memory untidied. +// +// An untidied local feed is hashed again from the bytes read, so the stored +// hash matches the stored data even if the file changed since it was fetched. +// A tidied local feed keeps the fetched hash; if the file changed in between, +// the next reload sees a new hash and imports it again. +func (f *feedFile) read(ctx context.Context, config Config) ([]byte, string, error) { + if config.EnableGTFSTidy { + logger := slog.Default().With(slog.String("component", "gtfs_loader")) + logging.LogOperation(logger, "gtfstidy_enabled_processing_gtfs_data") + tidiedData, err := tidyGTFSFile(ctx, f.path, logger) + if err == nil { + return tidiedData, f.hash, nil + } + logging.LogError(logger, "Failed to tidy GTFS data, using original data", err) + } + + b, err := os.ReadFile(f.path) + if err != nil { + return nil, "", fmt.Errorf("error reading GTFS data: %w", err) + } + if f.private { + return b, f.hash, nil + } + sum := sha256.Sum256(b) + return b, feedHash(hex.EncodeToString(sum[:]), config), nil +} diff --git a/internal/gtfs/feed_file_test.go b/internal/gtfs/feed_file_test.go new file mode 100644 index 000000000..f133c6195 --- /dev/null +++ b/internal/gtfs/feed_file_test.go @@ -0,0 +1,260 @@ +package gtfs + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "runtime" + "strconv" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "maglev.onebusaway.org/internal/appconf" + "maglev.onebusaway.org/internal/models" +) + +func rabaContentHash(t *testing.T) string { + t.Helper() + b, err := os.ReadFile(models.GetFixturePath(t, "raba.zip")) + require.NoError(t, err) + sum := sha256.Sum256(b) + return hex.EncodeToString(sum[:]) +} + +func TestFetchFeedFileLocal(t *testing.T) { + path := models.GetFixturePath(t, "raba.zip") + feed, err := fetchFeedFile(context.Background(), Config{GtfsURL: path}) + require.NoError(t, err) + defer feed.cleanup() + + assert.Equal(t, path, feed.path, "a local feed is read where it is") + assert.Equal(t, rabaContentHash(t), feed.hash) +} + +func TestFetchFeedFileDownload(t *testing.T) { + raba, err := os.ReadFile(models.GetFixturePath(t, "raba.zip")) + require.NoError(t, err) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("X-Api-Key") != "secret" { + w.WriteHeader(http.StatusForbidden) + return + } + _, _ = w.Write(raba) + })) + t.Cleanup(server.Close) + + dbDir := t.TempDir() + config := Config{ + GtfsURL: server.URL + "/gtfs.zip", + GTFSDataPath: filepath.Join(dbDir, "gtfs.db"), + StaticAuthHeaderKey: "X-Api-Key", + StaticAuthHeaderValue: "secret", + } + feed, err := fetchFeedFile(context.Background(), config) + require.NoError(t, err) + + assert.Equal(t, dbDir, filepath.Dir(feed.path), "the download goes next to the database") + assert.Equal(t, rabaContentHash(t), feed.hash) + feed.cleanup() + _, err = os.Stat(feed.path) + assert.True(t, os.IsNotExist(err), "cleanup removes the download") + + config.StaticAuthHeaderValue = "wrong" + _, err = fetchFeedFile(context.Background(), config) + assert.ErrorContains(t, err, "403") + entries, err := os.ReadDir(dbDir) + require.NoError(t, err) + assert.Empty(t, entries, "a failed download leaves no file behind") +} + +func TestFeedHashIncludesGTFSTidy(t *testing.T) { + assert.Equal(t, "abc", feedHash("abc", Config{})) + assert.Equal(t, "abc+gtfstidy", feedHash("abc", Config{EnableGTFSTidy: true})) +} + +func TestFetchFeedFileTruncatedDownloadLeavesNoFile(t *testing.T) { + const promisedBytes, sentBytes = 1024, 10 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Length", strconv.Itoa(promisedBytes)) + _, _ = w.Write(make([]byte, sentBytes)) + })) + t.Cleanup(server.Close) + + dbDir := t.TempDir() + _, err := fetchFeedFile(context.Background(), Config{GtfsURL: server.URL, GTFSDataPath: filepath.Join(dbDir, "gtfs.db")}) + assert.ErrorContains(t, err, "error reading GTFS data") + + entries, err := os.ReadDir(dbDir) + require.NoError(t, err) + assert.Empty(t, entries, "a truncated download leaves no file behind") +} + +func TestFetchFeedFileServerUnreachable(t *testing.T) { + server := httptest.NewServer(http.NotFoundHandler()) + server.Close() + + _, err := fetchFeedFile(context.Background(), Config{GtfsURL: server.URL, GTFSDataPath: filepath.Join(t.TempDir(), "gtfs.db")}) + assert.ErrorContains(t, err, "error downloading GTFS data") +} + +func TestNewDownloadLocation(t *testing.T) { + inMemory := downloadLocation{dir: os.TempDir(), prefix: "gtfs-static-pid" + strconv.Itoa(os.Getpid()) + "-"} + tests := []struct { + name, dataPath string + want downloadLocation + }{ + {"no database path", "", inMemory}, + {"in-memory database", ":memory:", inMemory}, + {"shared in-memory URI", "file::memory:?cache=shared", inMemory}, + {"in-memory mode URI", "file:gtfs.db?mode=memory", inMemory}, + {"database file", "/data/gtfs.db", downloadLocation{"/data", "gtfs-static-gtfs.db-"}}, + {"relative database file", "gtfs.db", downloadLocation{".", "gtfs-static-gtfs.db-"}}, + {"database URI", "file:/data/gtfs.db?cache=shared", downloadLocation{"/data", "gtfs-static-gtfs.db-"}}, + {"relative database URI", "file:data/gtfs.db", downloadLocation{"data", "gtfs-static-gtfs.db-"}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, newDownloadLocation(Config{GTFSDataPath: tt.dataPath})) + }) + } +} + +func TestDownloadLocationOwns(t *testing.T) { + location := downloadLocation{dir: ".", prefix: "gtfs-static-gtfs.db-"} + tests := []struct { + name string + want bool + }{ + {"gtfs-static-gtfs.db-123.zip", true}, + {"gtfs-static-gtfs.db-.zip", false}, + {"gtfs-static-gtfs.db-old-123.zip", false}, + {"gtfs-static-other.db-123.zip", false}, + {"gtfs-static-gtfs.db-123.zip.part", false}, + {"gtfs-static-123.zip", false}, + } + for _, tt := range tests { + assert.Equal(t, tt.want, location.owns(tt.name), tt.name) + } +} + +// installFakeGTFSTidy puts a gtfstidy on PATH that copies its input to its +// output, so tidying succeeds without the real binary. +func installFakeGTFSTidy(t *testing.T) { + t.Helper() + if runtime.GOOS == "windows" { + t.Skip("the fake gtfstidy is a shell script") + } + dir := t.TempDir() + // tidyGTFSData runs: gtfstidy -o + script := "#!/bin/sh\n/bin/cp \"$4\" \"$3\"\n" + require.NoError(t, os.WriteFile(filepath.Join(dir, "gtfstidy"), []byte(script), 0o755)) + t.Setenv("PATH", dir) +} + +// TestParseHashMarksTidyWhenItRan checks that a feed whose tidy succeeded is +// stored under the tidied hash that the next reload looks for. +func TestParseHashMarksTidyWhenItRan(t *testing.T) { + installFakeGTFSTidy(t) + config := Config{GtfsURL: models.GetFixturePath(t, "raba.zip"), EnableGTFSTidy: true} + + feed, err := fetchFeedFile(context.Background(), config) + require.NoError(t, err) + defer feed.cleanup() + + data, err := feed.parse(context.Background(), config) + require.NoError(t, err) + assert.Equal(t, rabaContentHash(t)+"+gtfstidy", data.Hash) + assert.Equal(t, feed.hash, data.Hash, "the stored hash matches the one the next reload checks") +} + +// TestReloadStaticSkipsParseForImportedFeed checks that reloading a feed the +// database already holds doesn't parse it again. +func TestReloadStaticSkipsParseForImportedFeed(t *testing.T) { + counter := countLogMessages(t) + + manager, err := InitGTFSManager(context.Background(), Config{ + GtfsURL: models.GetFixturePath(t, "raba.zip"), + GTFSDataPath: filepath.Join(t.TempDir(), "gtfs.db"), + Env: appconf.Development, + }) + require.NoError(t, err) + t.Cleanup(func() { _ = manager.Shutdown(context.Background()) }) + require.Equal(t, 1, counter.count("parsing_gtfs_data")) + + changed, err := manager.ReloadStatic(context.Background()) + require.NoError(t, err) + assert.False(t, changed) + assert.Equal(t, 1, counter.count("parsing_gtfs_data"), "an unchanged feed must not be parsed again") +} + +func TestFetchFeedFileRemovesStaleDownloads(t *testing.T) { + raba, err := os.ReadFile(models.GetFixturePath(t, "raba.zip")) + require.NoError(t, err) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write(raba) + })) + t.Cleanup(server.Close) + + dbDir := t.TempDir() + stale := filepath.Join(dbDir, "gtfs-static-gtfs.db-123.zip") + require.NoError(t, os.WriteFile(stale, []byte("left by a crash"), 0o600)) + othersDownload := filepath.Join(dbDir, "gtfs-static-other.db-456.zip") + require.NoError(t, os.WriteFile(othersDownload, []byte("in use by another process"), 0o600)) + + feed, err := fetchFeedFile(context.Background(), Config{GtfsURL: server.URL, GTFSDataPath: filepath.Join(dbDir, "gtfs.db")}) + require.NoError(t, err) + defer feed.cleanup() + _, err = os.Stat(stale) + assert.True(t, os.IsNotExist(err), "a download left by this database's process is removed") + assert.FileExists(t, othersDownload, "another database's download is left alone") +} + +// TestReloadStaticSkipsParseWhenTidyFails checks that a feed whose tidy fails +// is stored under the hash the next reload looks for, so it isn't parsed again. +func TestReloadStaticSkipsParseWhenTidyFails(t *testing.T) { + t.Setenv("PATH", t.TempDir()) // no gtfstidy binary, so tidying fails + counter := countLogMessages(t) + + manager, err := InitGTFSManager(context.Background(), Config{ + GtfsURL: models.GetFixturePath(t, "raba.zip"), + GTFSDataPath: filepath.Join(t.TempDir(), "gtfs.db"), + Env: appconf.Development, + EnableGTFSTidy: true, + }) + require.NoError(t, err) + t.Cleanup(func() { _ = manager.Shutdown(context.Background()) }) + require.Equal(t, 1, counter.count("parsing_gtfs_data")) + + changed, err := manager.ReloadStatic(context.Background()) + require.NoError(t, err) + assert.False(t, changed) + assert.Equal(t, 1, counter.count("parsing_gtfs_data"), "a feed whose tidy failed must not be parsed again") +} + +// TestParseHashesOnlyLocalFeedsAgain checks that parse trusts the hash of a +// download, which nothing else can change, and hashes a local feed again. +func TestParseHashesOnlyLocalFeedsAgain(t *testing.T) { + const fetchedHash = "hash-when-fetched" + path := models.GetFixturePath(t, "raba.zip") + tests := []struct { + name string + private bool + want string + }{ + {"download", true, fetchedHash}, + {"local feed", false, rabaContentHash(t)}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + feed := &feedFile{path: path, hash: fetchedHash, private: tt.private, cleanup: func() {}} + data, err := feed.parse(context.Background(), Config{GtfsURL: path}) + require.NoError(t, err) + assert.Equal(t, tt.want, data.Hash) + }) + } +} diff --git a/internal/gtfs/realtime.go b/internal/gtfs/realtime.go index 5f4c64da5..c51163c49 100644 --- a/internal/gtfs/realtime.go +++ b/internal/gtfs/realtime.go @@ -159,12 +159,7 @@ const maxVehicleFeedFutureSkew = 15 * time.Minute var realtimeHTTPClient = newRealtimeHTTPClient() func newRealtimeHTTPClient() *http.Client { - var transport *http.Transport - if t, ok := http.DefaultTransport.(*http.Transport); ok { - transport = t.Clone() - } else { - transport = &http.Transport{} - } + transport := cloneDefaultTransport() transport.MaxIdleConns = 50 transport.MaxIdleConnsPerHost = 10 transport.IdleConnTimeout = 90 * time.Second diff --git a/internal/gtfs/static.go b/internal/gtfs/static.go index 4371cf031..f6c51ee4a 100644 --- a/internal/gtfs/static.go +++ b/internal/gtfs/static.go @@ -3,10 +3,8 @@ package gtfs import ( "context" "fmt" - "io" "log/slog" - "net/http" - "os" + "runtime/debug" "strings" "time" @@ -15,74 +13,8 @@ import ( "maglev.onebusaway.org/internal/logging" ) -func rawGtfsData(ctx context.Context, source string, config Config) ([]byte, error) { - var b []byte - var err error - - logger := slog.Default().With(slog.String("component", "gtfs_loader")) - logging.LogOperation(logger, "reading_gtfs_data", slog.String("source", source)) - - if config.isLocalFile() { - b, err = os.ReadFile(source) - if err != nil { - return nil, fmt.Errorf("error reading local GTFS file: %w", err) - } - } else { - req, err := http.NewRequestWithContext(ctx, "GET", source, nil) - if err != nil { - return nil, fmt.Errorf("error creating GTFS request: %w", err) - } - - // Add auth header if provided - if config.StaticAuthHeaderKey != "" && config.StaticAuthHeaderValue != "" { - req.Header.Set(config.StaticAuthHeaderKey, config.StaticAuthHeaderValue) - } - - client := &http.Client{ - Timeout: 5 * time.Minute, - Transport: &http.Transport{ - TLSHandshakeTimeout: 10 * time.Second, - ResponseHeaderTimeout: 30 * time.Second, - IdleConnTimeout: 90 * time.Second, - }} - - resp, err := client.Do(req) - if err != nil { - return nil, fmt.Errorf("error downloading GTFS data: %w", err) - } - defer logging.SafeCloseWithLogging(resp.Body, - slog.Default().With(slog.String("component", "gtfs_downloader")), - "http_response_body") - - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("failed to download GTFS data: received HTTP status %s", resp.Status) - } - const maxStaticSize = 200 * 1024 * 1024 - b, err = io.ReadAll(io.LimitReader(resp.Body, maxStaticSize+1)) - if err != nil { - return nil, fmt.Errorf("error reading GTFS data: %w", err) - } - if int64(len(b)) > maxStaticSize { - return nil, fmt.Errorf("static GTFS response exceeds size limit of %d bytes", maxStaticSize) - } - } - - // Process through gtfstidy if enabled - if config.EnableGTFSTidy { - logging.LogOperation(logger, "gtfstidy_enabled_processing_gtfs_data") - tidiedData, err := tidyGTFSData(ctx, b, logger) - if err != nil { - logging.LogError(logger, "Failed to tidy GTFS data, using original data", err) - } else { - b = tidiedData - } - } - - return b, nil -} - // openGtfsDB opens (or creates) the single SQLite database used by the manager. -// No import work happens here — use importStaticIntoDB against the returned client. +// No import work happens here — ReloadStatic imports into the returned client. func openGtfsDB(config Config) (*gtfsdb.Client, error) { dbConfig := newGTFSDBConfig(config.GTFSDataPath, config) client, err := gtfsdb.NewClient(dbConfig) @@ -92,28 +24,15 @@ func openGtfsDB(config Config) (*gtfsdb.Client, error) { return client, nil } -// importStaticIntoDB imports the already-parsed GTFS data into the provided client. -// Returns (changed, err): changed is true when the DB was actually updated. When changed, -// it also precomputes stop directions. Trip time bounds are now computed inside the -// import transaction by ImportParsedGTFS itself. -func importStaticIntoDB(ctx context.Context, client *gtfsdb.Client, data *gtfsdb.GtfsData) (bool, error) { - changed, err := client.StoreGtfsData(ctx, data) - if err != nil { - return false, err - } - - if !changed { - return false, nil - } - - logger := slog.Default().With(slog.String("component", "gtfs_db_builder")) +// precomputeStopDirections stores each stop's direction for the feed now in +// the database. A failure is logged, not returned: the API computes missing +// directions on demand. +func precomputeStopDirections(ctx context.Context, client *gtfsdb.Client) { precomputer := NewDirectionPrecomputer(client.Queries, client.DB) if err := precomputer.PrecomputeAllDirections(ctx); err != nil { - // Log error but don't fail the entire import + logger := slog.Default().With(slog.String("component", "gtfs_db_builder")) logging.LogError(logger, "Failed to precompute stop directions - API will fallback to on-demand calculation", err) } - - return true, nil } func newGTFSDBConfig(dbPath string, config Config) gtfsdb.Config { @@ -124,25 +43,6 @@ func newGTFSDBConfig(dbPath string, config Config) gtfsdb.Config { return dbConfig } -// loadGTFSData loads, parses, hashes, and validates GTFS data from either a URL or a local file. -func loadGTFSData(ctx context.Context, config Config) (*gtfsdb.GtfsData, error) { - b, err := rawGtfsData(ctx, config.GtfsURL, config) - if err != nil { - return nil, fmt.Errorf("error reading GTFS data: %w", err) - } - - data, err := gtfsdb.ParseGtfsData(b, config.GtfsURL) - if err != nil { - return nil, err - } - - if err := validateStaticAgencyTimezones(data.Static); err != nil { - return nil, fmt.Errorf("invalid GTFS agency timezone: %w", err) - } - - return data, nil -} - func validateStaticAgencyTimezones(staticData *gtfs.Static) error { for i, agency := range staticData.Agencies { tz := strings.TrimSpace(agency.Timezone) @@ -199,26 +99,20 @@ func (manager *Manager) ReloadStatic(ctx context.Context) (bool, error) { defer manager.staticUpdateMutex.Unlock() logger := slog.Default().With(slog.String("component", "gtfs_updater")) - newData, err := loadGTFSData(ctx, manager.config) - if err != nil { - logging.LogError(logger, "Error loading GTFS data", err, - slog.String("source", manager.config.GtfsURL)) - return false, err - } - - if err := ctx.Err(); err != nil { - return false, err - } - - changed, err := importStaticIntoDB(ctx, manager.GtfsDB, newData) + changed, err := manager.storeConfiguredFeed(ctx, logger) if err != nil { - logging.LogError(logger, "Error importing GTFS data", err) return false, err } - if !changed { + if changed { + // The parsed feed became garbage when storeConfiguredFeed returned. + // Collecting it now, and returning its pages to the OS, keeps it out + // of the precompute's memory and out of the process's footprint. + debug.FreeOSMemory() + precomputeStopDirections(ctx, manager.GtfsDB) + } else { logging.LogOperation(logger, "gtfs_static_data_unchanged", - slog.String("source", newData.Source)) + slog.String("source", manager.config.GtfsURL)) } newRegionBounds := computeRegionBounds(ctx, manager.GtfsDB) @@ -249,6 +143,48 @@ func (manager *Manager) ReloadStatic(ctx context.Context) (bool, error) { return changed, nil } +// storeConfiguredFeed fetches the configured feed and, unless the database +// already holds it, parses and stores it. The parsed feed is reachable only +// from here, so it can be freed as soon as this returns. It reports whether +// the database changed. +func (manager *Manager) storeConfiguredFeed(ctx context.Context, logger *slog.Logger) (bool, error) { + feed, err := fetchFeedFile(ctx, manager.config) + if err != nil { + logging.LogError(logger, "Error loading GTFS data", err, + slog.String("source", manager.config.GtfsURL)) + return false, err + } + defer feed.cleanup() + + skipped, err := manager.GtfsDB.SkipIfImported(ctx, feed.hash, manager.config.GtfsURL) + if err != nil { + logging.LogError(logger, "Error checking whether GTFS data is already imported", err, + slog.String("source", manager.config.GtfsURL)) + return false, err + } + if skipped { + return false, nil + } + + newData, err := feed.parse(ctx, manager.config) + if err != nil { + logging.LogError(logger, "Error loading GTFS data", err, + slog.String("source", manager.config.GtfsURL)) + return false, err + } + + if err := ctx.Err(); err != nil { + return false, err + } + + changed, err := manager.GtfsDB.StoreGtfsData(ctx, newData) + if err != nil { + logging.LogError(logger, "Error importing GTFS data", err) + return false, err + } + return changed, nil +} + // logFeedExpiry reads the feed_expires_at value persisted by StoreGtfsData // and updates the metrics gauge / emits warning logs about how soon the feed // will expire. The DB write itself happens atomically inside the import diff --git a/internal/gtfs/static_test.go b/internal/gtfs/static_test.go index 2b1507ddf..51d6c1c2d 100644 --- a/internal/gtfs/static_test.go +++ b/internal/gtfs/static_test.go @@ -1,11 +1,19 @@ package gtfs import ( + "context" + "database/sql" + "log/slog" + "path/filepath" + "sync" "testing" + "time" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "maglev.onebusaway.org/internal/appconf" "maglev.onebusaway.org/internal/metrics" + "maglev.onebusaway.org/internal/models" ) func TestNewGTFSDBConfig_QueryMetricsRecorder(t *testing.T) { @@ -27,3 +35,128 @@ func TestNewGTFSDBConfig_QueryMetricsRecorder(t *testing.T) { assert.Same(t, m, dbConfig.QueryMetricsRecorder) }) } + +// logMessageCounter is a slog.Handler that counts records by message. +type logMessageCounter struct { + mu sync.Mutex + counts map[string]int +} + +func (c *logMessageCounter) Enabled(context.Context, slog.Level) bool { return true } +func (c *logMessageCounter) WithAttrs([]slog.Attr) slog.Handler { return c } +func (c *logMessageCounter) WithGroup(string) slog.Handler { return c } +func (c *logMessageCounter) Handle(_ context.Context, record slog.Record) error { + c.mu.Lock() + defer c.mu.Unlock() + c.counts[record.Message]++ + return nil +} + +func (c *logMessageCounter) count(message string) int { + c.mu.Lock() + defer c.mu.Unlock() + return c.counts[message] +} + +// countLogMessages routes the default logger to a counter for the rest of the +// test. +func countLogMessages(t *testing.T) *logMessageCounter { + counter := &logMessageCounter{counts: map[string]int{}} + previous := slog.Default() + slog.SetDefault(slog.New(counter)) + t.Cleanup(func() { slog.SetDefault(previous) }) + return counter +} + +// TestReloadStaticPrecomputesOnlyWhenChanged checks that stop directions are +// precomputed after an import that changed the data, and not after a reload +// that found the same feed. +func TestReloadStaticPrecomputesOnlyWhenChanged(t *testing.T) { + counter := countLogMessages(t) + manager, err := InitGTFSManager(context.Background(), Config{ + GtfsURL: models.GetFixturePath(t, "raba.zip"), + GTFSDataPath: filepath.Join(t.TempDir(), "gtfs.db"), + Env: appconf.Development, + }) + require.NoError(t, err) + t.Cleanup(func() { _ = manager.Shutdown(context.Background()) }) + require.Equal(t, 1, counter.count("precomputing_stop_directions_started")) + + var withDirection int + require.NoError(t, manager.GtfsDB.DB.QueryRow(`SELECT COUNT(*) FROM stops WHERE direction IS NOT NULL`).Scan(&withDirection)) + assert.Positive(t, withDirection) + + changed, err := manager.ReloadStatic(context.Background()) + require.NoError(t, err) + assert.False(t, changed) + assert.Equal(t, 1, counter.count("precomputing_stop_directions_started")) +} + +// TestReloadStaticUnchangedFeedRefreshesExpiry checks that a reload that finds +// the same feed skips the parse but still recomputes feed_expires_at, so a +// database written by an older build picks up a corrected expiry. +func TestReloadStaticUnchangedFeedRefreshesExpiry(t *testing.T) { + counter := countLogMessages(t) + manager, err := InitGTFSManager(context.Background(), Config{ + GtfsURL: models.GetFixturePath(t, "raba.zip"), + GTFSDataPath: filepath.Join(t.TempDir(), "gtfs.db"), + Env: appconf.Development, + }) + require.NoError(t, err) + t.Cleanup(func() { _ = manager.Shutdown(context.Background()) }) + + want := manager.FeedExpiresAt(context.Background()) + require.False(t, want.IsZero()) + stale := sql.NullInt64{Int64: want.Add(-24 * time.Hour).Unix(), Valid: true} + require.NoError(t, manager.GtfsDB.Queries.UpdateFeedExpiresAt(context.Background(), stale)) + + changed, err := manager.ReloadStatic(context.Background()) + require.NoError(t, err) + assert.False(t, changed) + assert.Equal(t, 1, counter.count("gtfs_data_unchanged_skipping_import")) + assert.Equal(t, want, manager.FeedExpiresAt(context.Background())) +} + +// initManagerWithFileDB imports RABA into a database file and returns the +// manager, so a test can then change the database under a reload. +func initManagerWithFileDB(t *testing.T) *Manager { + t.Helper() + manager, err := InitGTFSManager(context.Background(), Config{ + GtfsURL: models.GetFixturePath(t, "raba.zip"), + GTFSDataPath: filepath.Join(t.TempDir(), "gtfs.db"), + Env: appconf.Development, + }) + require.NoError(t, err) + t.Cleanup(func() { _ = manager.Shutdown(context.Background()) }) + return manager +} + +// TestReloadStaticUnchangedFeedExpiryFailure checks that a reload of an +// unchanged feed reports a failure to refresh feed_expires_at, and doesn't +// fall back to parsing the feed. +func TestReloadStaticUnchangedFeedExpiryFailure(t *testing.T) { + counter := countLogMessages(t) + manager := initManagerWithFileDB(t) + require.Equal(t, 1, counter.count("parsing_gtfs_data")) + + _, err := manager.GtfsDB.DB.Exec(`UPDATE agencies SET timezone = 'Invalid/Timezone'`) + require.NoError(t, err) + + changed, err := manager.ReloadStatic(context.Background()) + assert.ErrorContains(t, err, "failed to refresh feed expiry") + assert.False(t, changed) + assert.Equal(t, 1, counter.count("parsing_gtfs_data"), "a failed refresh must not parse the feed") +} + +// TestReloadStaticSkipIfImportedError checks that a reload reports a failure to +// read import_metadata instead of treating the feed as new. +func TestReloadStaticSkipIfImportedError(t *testing.T) { + manager := initManagerWithFileDB(t) + + _, err := manager.GtfsDB.DB.Exec(`DROP TABLE import_metadata`) + require.NoError(t, err) + + changed, err := manager.ReloadStatic(context.Background()) + assert.ErrorContains(t, err, "error checking import metadata") + assert.False(t, changed) +} diff --git a/internal/gtfs/tidy.go b/internal/gtfs/tidy.go index b61970beb..a9c0bbdff 100644 --- a/internal/gtfs/tidy.go +++ b/internal/gtfs/tidy.go @@ -50,13 +50,23 @@ func checkGTFSTidyAvailable() (string, error) { return path, nil } -// tidyGTFSData processes GTFS data through gtfstidy with flags -OscRCSmeD -// Returns the tidied GTFS zip data or an error -func tidyGTFSData(ctx context.Context, inputZip []byte, logger *slog.Logger) ([]byte, error) { +// tidyGTFSFile processes the GTFS zip at inputPath through gtfstidy with +// flags -OscRCSmeD. gtfstidy reads the file itself, so the untidied feed is +// never loaded into memory. Returns the tidied GTFS zip data or an error. +func tidyGTFSFile(ctx context.Context, inputPath string, logger *slog.Logger) ([]byte, error) { gtfstidyPath, err := checkGTFSTidyAvailable() if err != nil { return nil, err } + // gtfstidy runs in its own directory, so a relative path would miss. + inputPath, err = filepath.Abs(inputPath) + if err != nil { + return nil, fmt.Errorf("failed to resolve pristine GTFS zip path: %w", err) + } + input, err := os.Stat(inputPath) + if err != nil { + return nil, fmt.Errorf("failed to read pristine GTFS zip: %w", err) + } // Create temp directory for processing // Note: We should clean up the temp dir on error || after processing @@ -71,19 +81,14 @@ func tidyGTFSData(ctx context.Context, inputZip []byte, logger *slog.Logger) ([] } }() - pristineZipPath := filepath.Join(tempDir, "gtfs_pristine.zip") - if err := os.WriteFile(pristineZipPath, inputZip, 0600); err != nil { - return nil, fmt.Errorf("failed to write pristine GTFS zip: %w", err) - } - outputZipPath := filepath.Join(tempDir, "gtfs_tidied.zip") logging.LogOperation(logger, "running_gtfstidy", - slog.String("input_file", pristineZipPath), - slog.Int("input_size_bytes", len(inputZip))) + slog.String("input_file", inputPath), + slog.Int64("input_size_bytes", input.Size())) // -o: output file - cmd := exec.CommandContext(ctx, gtfstidyPath, "-OscRCSmeD", "-o", outputZipPath, pristineZipPath) + cmd := exec.CommandContext(ctx, gtfstidyPath, "-OscRCSmeD", "-o", outputZipPath, inputPath) cmd.Dir = tempDir var stdout, stderr bytes.Buffer @@ -107,7 +112,7 @@ func tidyGTFSData(ctx context.Context, inputZip []byte, logger *slog.Logger) ([] logging.LogOperation(logger, "gtfs_tidied_successfully", slog.Int("output_size_bytes", len(tidiedZip)), - slog.Int("size_reduction_bytes", len(inputZip)-len(tidiedZip))) + slog.Int64("size_reduction_bytes", input.Size()-int64(len(tidiedZip)))) return tidiedZip, nil } diff --git a/internal/gtfs/tidy_test.go b/internal/gtfs/tidy_test.go index 308ff61fd..53aa051c9 100644 --- a/internal/gtfs/tidy_test.go +++ b/internal/gtfs/tidy_test.go @@ -10,7 +10,7 @@ import ( "testing" ) -func TestTidyGTFSData(t *testing.T) { +func TestTidyGTFSFile(t *testing.T) { projectRoot, err := findProjectRoot() if err != nil { @@ -38,9 +38,9 @@ func TestTidyGTFSData(t *testing.T) { logger := slog.New(slog.NewTextHandler(os.Stdout, nil)) // Run tidy - outputData, err := tidyGTFSData(context.Background(), inputData, logger) + outputData, err := tidyGTFSFile(context.Background(), testDataPath, logger) if err != nil { - t.Fatalf("tidyGTFSData failed: %v", err) + t.Fatalf("tidyGTFSFile failed: %v", err) } // Verify output is not empty