Stream tarballs file-by-file instead of extracting to disk

Add ExtractFromTarGz to stream tar entries and parse only JS/TS
files one at a time, avoiding writing the full tarball to Cloud
Run's memory-backed filesystem. The service now pipes the HTTP
response body directly into the streaming extractor.

Also bump instance to 4 vCPU / 16Gi with scale-to-zero and
concurrency=20, and use clog/gcp/init for structured logging.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

cmd/ast-diff-service/main.go

@@ -4,10 +4,10 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"os"
 	"strings"
 
 	"github.com/chainguard-dev/clog"
+	_ "github.com/chainguard-dev/clog/gcp/init"
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	cehttp "github.com/cloudevents/sdk-go/v2/protocol/http"
 	"github.com/sethvargo/go-envconfig"
@@ -114,38 +114,24 @@ func handleRelease(ctx context.Context, sender cloudevents.Client, event cloudev
 	oldURL := npm.TarballURL(rel.Name, oldVersion)
 	newURL := npm.TarballURL(rel.Name, rel.Version)
 
-	oldDir, err := npm.DownloadAndExtract(oldURL)
+	oldSyms, oldHashes, err := downloadAndExtractSymbols(oldURL)
 	if err != nil {
 		if npm.IsPermanent(err) {
 			log.InfoContext(ctx, "skipping old version", "error", err)
 			return nil
 		}
-		log.ErrorContext(ctx, "failed to download old version", "error", err)
-		return fmt.Errorf("downloading old version: %w", err)
+		log.ErrorContext(ctx, "failed to process old version", "error", err)
+		return fmt.Errorf("processing old version: %w", err)
 	}
-	defer os.RemoveAll(oldDir)
 
-	newDir, err := npm.DownloadAndExtract(newURL)
+	newSyms, newHashes, err := downloadAndExtractSymbols(newURL)
 	if err != nil {
 		if npm.IsPermanent(err) {
 			log.InfoContext(ctx, "skipping new version", "error", err)
 			return nil
 		}
-		log.ErrorContext(ctx, "failed to download new version", "error", err)
-		return fmt.Errorf("downloading new version: %w", err)
-	}
-	defer os.RemoveAll(newDir)
-
-	oldSyms, oldHashes, err := symbols.ExtractFromDir(oldDir)
-	if err != nil {
-		log.ErrorContext(ctx, "failed to extract old symbols", "error", err)
-		return nil // parse failures are not retryable
-	}
-
-	newSyms, newHashes, err := symbols.ExtractFromDir(newDir)
-	if err != nil {
-		log.ErrorContext(ctx, "failed to extract new symbols", "error", err)
-		return nil // parse failures are not retryable
+		log.ErrorContext(ctx, "failed to process new version", "error", err)
+		return fmt.Errorf("processing new version: %w", err)
 	}
 
 	added, removed, modified := diff.Diff(oldSyms, oldHashes, newSyms, newHashes)
@@ -191,6 +177,18 @@ func handleRelease(ctx context.Context, sender cloudevents.Client, event cloudev
 	return nil
 }
 
+// downloadAndExtractSymbols streams a tarball and extracts symbols
+// from JS/TS files without writing to disk.
+func downloadAndExtractSymbols(url string) (map[string]symbols.Symbol, symbols.SymbolMap, error) {
+	body, err := npm.DownloadTarball(url)
+	if err != nil {
+		return nil, nil, err
+	}
+	defer body.Close()
+
+	return symbols.ExtractFromTarGz(body)
+}
+
 func mustJSON(v any) string {
 	b, err := json.Marshal(v)
 	if err != nil {

internal/npm/npm.go

@@ -145,6 +145,26 @@ func DownloadAndExtract(url string) (string, error) {
 	return dir, nil
 }
 
+// DownloadTarball downloads a tarball and returns the response body for
+// streaming. The caller must close the returned ReadCloser.
+func DownloadTarball(url string) (io.ReadCloser, error) {
+	resp, err := http.Get(url)
+	if err != nil {
+		return nil, fmt.Errorf("downloading %s: %w", url, err)
+	}
+
+	if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusForbidden {
+		resp.Body.Close()
+		return nil, &PermanentError{Err: fmt.Errorf("status %d for %s", resp.StatusCode, url)}
+	}
+	if resp.StatusCode != http.StatusOK {
+		resp.Body.Close()
+		return nil, fmt.Errorf("unexpected status %d for %s", resp.StatusCode, url)
+	}
+
+	return resp.Body, nil
+}
+
 const maxFileSize = 10 * 1024 * 1024 // 10MB per file
 
 func extractTarGz(r io.Reader, dest string) error {

internal/symbols/symbols.go

@@ -1,8 +1,11 @@
 package symbols
 
 import (
+	"archive/tar"
+	"compress/gzip"
 	"crypto/sha256"
 	"fmt"
+	"io"
 	"os"
 	"path/filepath"
 	"strings"
@@ -84,6 +87,81 @@ func ExtractFromDir(dir string) (map[string]Symbol, SymbolMap, error) {
 	return symbols, hashes, err
 }
 
+const maxFileSize = 10 * 1024 * 1024 // 10MB per file
+
+// ExtractFromTarGz streams a tar.gz reader and extracts symbols from
+// JS/TS files one at a time, without writing anything to disk.
+func ExtractFromTarGz(r io.Reader) (map[string]Symbol, SymbolMap, error) {
+	syms := make(map[string]Symbol)
+	hashes := make(SymbolMap)
+
+	jsLang := grammars.JavascriptLanguage()
+	tsLang := grammars.TypescriptLanguage()
+
+	gz, err := gzip.NewReader(r)
+	if err != nil {
+		return nil, nil, err
+	}
+	defer gz.Close()
+
+	tr := tar.NewReader(gz)
+	for {
+		hdr, err := tr.Next()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			return nil, nil, err
+		}
+		if hdr.Typeflag != tar.TypeReg || hdr.Size > maxFileSize {
+			continue
+		}
+
+		// Strip the "package/" prefix npm tarballs use.
+		name := strings.TrimPrefix(hdr.Name, "package/")
+		if name == "" || strings.Contains(name, "..") {
+			continue
+		}
+
+		// Skip node_modules and .git directories.
+		if strings.HasPrefix(name, "node_modules/") || strings.Contains(name, "/node_modules/") {
+			continue
+		}
+
+		ext := filepath.Ext(name)
+		var lang *gotreesitter.Language
+		switch ext {
+		case ".js", ".mjs", ".cjs":
+			lang = jsLang
+		case ".ts", ".mts", ".cts":
+			lang = tsLang
+		default:
+			continue
+		}
+
+		if strings.HasSuffix(strings.TrimSuffix(name, ext), ".min") {
+			continue
+		}
+
+		src, err := io.ReadAll(io.LimitReader(tr, maxFileSize))
+		if err != nil {
+			continue
+		}
+
+		extracted, err := extractSymbols(src, lang, name)
+		if err != nil {
+			continue
+		}
+
+		for _, sym := range extracted {
+			syms[sym.sym.Key()] = sym.sym
+			hashes[sym.sym.Key()] = sym.hash
+		}
+	}
+
+	return syms, hashes, nil
+}
+
 type extractedSymbol struct {
 	sym  Symbol
 	hash string

lessons.md

@@ -0,0 +1,11 @@
+# Lessons
+
+- Don't seek/clear PubSub backlogs to test fixes — watch metrics naturally to see if the system stabilizes. Clearing the queue masks whether the fix actually works.
+- When setting container options in `regional-go-service`, `cpu_idle` and resource `limits` go inside `resources = {}`, not at the container top level.
+- CGO-based tree-sitter bindings can't cross-compile with `ko` for Linux containers. Use `gotreesitter` (pure Go, embedded grammars) instead.
+- Cloud Run services behind PubSub push subscriptions need `egress = "PRIVATE_RANGES_ONLY"` if they need to reach external APIs (like registry.npmjs.org). The default `ALL_TRAFFIC` routes everything through VPC, which can't reach the public internet.
+- Distinguish permanent errors (404, 403) from transient errors (timeouts, 5xx) in event handlers. Returning an error to PubSub triggers a retry — permanent failures should be ACKed so they don't waste the retry budget and end up in the DLQ.
+- For bursty event-driven services, prefer scale-to-zero with a beefy instance over always-on with a small instance. PubSub retries handle cold-start 503s naturally, and the bigger instance processes the burst faster. Much cheaper than keeping an instance warm 24/7.
+- Low concurrency settings (e.g. 10) cause Cloud Run to spin up many instances during a burst, each needing a cold start. Default concurrency (80) lets one instance absorb more of the burst.
+- `cpu-throttling: true` (cpu_idle) is fine for bursty workloads — the CPU gets allocated when a request arrives. Setting it to false keeps CPU allocated 24/7, which is expensive for a service that's idle most of the time.
+- Cloud Run's filesystem is backed by memory. Writing tarballs to disk (via temp dirs) consumes the container's memory limit, causing OOMs on large packages. Stream tar entries and only buffer individual JS/TS files — never write the full tarball to disk.

main.go

@@ -11,6 +11,7 @@ import (
 
 	"cloud.google.com/go/storage"
 	"github.com/chainguard-dev/clog"
+	_ "github.com/chainguard-dev/clog/gcp/init"
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	cehttp "github.com/cloudevents/sdk-go/v2/protocol/http"
 	"github.com/sethvargo/go-envconfig"

terraform/main.tf

@@ -195,6 +195,12 @@ module "ast-diff" {
         importpath  = "./cmd/ast-diff-service"
       }
       ports = [{ container_port = 8080 }]
+      resources = {
+        limits = {
+          cpu    = "4"
+          memory = "16Gi"
+        }
+      }
       env = [{
         name  = "K_SINK"
         value = module.authorize-ast-diff.uri
@@ -209,14 +215,6 @@ module "ast-diff" {
   # Allow more time for large packages.
   request_timeout_seconds = 300
 
-  # Keep one instance warm and limit concurrency so Cloud Run
-  # scales out earlier. Each diff downloads tarballs and parses
-  # files, so high concurrency per instance causes contention.
-  scaling = {
-    min_instances                    = 1
-    max_instance_request_concurrency = 10
-  }
-
   # PRIVATE_RANGES_ONLY so broker traffic routes through VPC
   # but npm registry traffic goes directly over the internet.
   egress = "PRIVATE_RANGES_ONLY"
@@ -262,12 +260,18 @@ module "cron-dashboard" {
 }
 
 module "ast-diff-dashboard" {
-  source = "chainguard-dev/common/infra//modules/dashboard/service"
+  source = "chainguard-dev/common/infra//modules/dashboard/cloudevent-receiver"
 
   project_id            = var.project_id
   service_name          = "npm-snoop-ast-diff"
   notification_channels = []
 
+  triggers = {
+    "released" = {
+      subscription_prefix = "npm-snoop-ast-diff"
+    }
+  }
+
   sections = {
     http = true
     grpc = false