feat: Phase 3 crowdsourced cache contribution

- Redis L2 persistent cache: initRedis() from REDIS_URL env, seed
  in-memory linkCache on startup, write-through on every fresh fetch
  and background refresh (silent fallback to memory-only if unavailable)
- POST /contribute endpoint: validates product catalog, expiry > 1h,
  Microsoft CDN host allow-list, per-IP token bucket (~5/min),
  CONTRIBUTE_SECRET header; writes to in-memory + Redis; logs all
  accepts and rejections
- CLI --no-contribute flag (also MSDL_NO_CONTRIBUTE=1 env var):
  background goroutine posts raw MS JSON to /contribute after every
  consumer fetch; best-effort 5s timeout; opt-out message to stderr

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Shekhar Vaidya 2026-06-18 14:07:35 +05:30
parent 7ebf5a9560
commit 9a5f41290d
5 changed files with 419 additions and 7 deletions

View file

@ -3,6 +3,9 @@ module msdl-backend-go
go 1.25.0
require (
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/redis/go-redis/v9 v9.20.1 // indirect
go.uber.org/atomic v1.11.0 // indirect
golang.org/x/sync v0.20.0 // indirect
)

View file

@ -1,4 +1,10 @@
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/redis/go-redis/v9 v9.20.1 h1:sfCU6A8P3dXbKyWes02uxA2baehGux9dZHfEKtsTB1w=
github.com/redis/go-redis/v9 v9.20.1/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA=
go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=

View file

@ -2,6 +2,7 @@ package main
import (
"bytes"
"context"
"encoding/json"
"fmt"
"html"
@ -20,6 +21,7 @@ import (
"time"
"github.com/google/uuid"
"github.com/redis/go-redis/v9"
"golang.org/x/sync/singleflight"
)
@ -131,6 +133,10 @@ var (
var sfGroup singleflight.Group
// --- Redis L2 cache client ---
var rdb *redis.Client
// --- Metrics counters (atomic — no mutex needed) ---
var (
@ -285,6 +291,198 @@ type rateLimitError struct{ message string }
func (e *rateLimitError) Error() string { return e.message }
// --- Redis L2 cache helpers ---
func redisKey(productID, skuID string) string {
return "msdl:link:" + productID + ":" + skuID
}
func initRedis() {
redisURL := os.Getenv("REDIS_URL")
if redisURL == "" {
log.Println("redis: REDIS_URL not set — memory-only mode")
return
}
opts, err := redis.ParseURL(redisURL)
if err != nil {
log.Printf("redis: invalid REDIS_URL: %v — memory-only mode\n", err)
return
}
client := redis.NewClient(opts)
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
if err := client.Ping(ctx).Err(); err != nil {
log.Printf("redis: ping failed: %v — memory-only mode\n", err)
return
}
rdb = client
log.Println("redis: connected")
}
func redisWriteLink(productID, skuID string, raw []byte, expiresAt time.Time) {
if rdb == nil {
return
}
ttl := time.Until(expiresAt)
if ttl <= 0 {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
if err := rdb.Set(ctx, redisKey(productID, skuID), raw, ttl).Err(); err != nil {
log.Printf("redis: write failed for %s:%s: %v\n", productID, skuID, err)
}
}
func redisSeedLinkCache() {
if rdb == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
var cursor uint64
var seeded int
for {
keys, nextCursor, err := rdb.Scan(ctx, cursor, "msdl:link:*", 100).Result()
if err != nil {
log.Printf("redis: seed scan error: %v\n", err)
return
}
for _, key := range keys {
raw, err := rdb.Get(ctx, key).Bytes()
if err != nil {
continue
}
ttl, err := rdb.TTL(ctx, key).Result()
if err != nil || ttl <= 0 {
continue
}
// Key format: "msdl:link:<productID>:<skuID>"
parts := strings.SplitN(strings.TrimPrefix(key, "msdl:link:"), ":", 2)
if len(parts) != 2 {
continue
}
cacheKey := parts[0] + ":" + parts[1]
expiresAt := time.Now().Add(ttl)
linkCacheMu.Lock()
linkCache[cacheKey] = linkCacheEntry{RawJSON: raw, ExpiresAt: expiresAt, FetchedAt: time.Now()}
linkCacheMu.Unlock()
seeded++
}
cursor = nextCursor
if cursor == 0 {
break
}
}
if seeded > 0 {
log.Printf("redis: seeded %d link cache entries from Redis\n", seeded)
}
}
// --- /contribute validation helpers ---
// validContributeProducts is the canonical product ID allow-list for POST /contribute.
var validContributeProducts = map[string]bool{
"52": true, "2378": true, "2618": true,
"3113": true, "3114": true, "3115": true,
"3131": true, "3132": true, "3133": true,
"3262": true, "3263": true, "3264": true,
"3265": true, "3266": true, "3267": true,
"3321": true, "3324": true,
}
// allowedCDNSuffixes is the CDN host allow-list for contributed download URLs.
// Worst-case abuse: attacker submits a valid Microsoft-signed link — harmless.
var allowedCDNSuffixes = []string{
".download.prss.microsoft.com",
".dl.delivery.mp.microsoft.com",
".delivery.mp.microsoft.com",
}
func isAllowedCDNHost(rawURL string) bool {
u, err := url.Parse(rawURL)
if err != nil {
return false
}
host := strings.ToLower(u.Hostname())
for _, suffix := range allowedCDNSuffixes {
if strings.HasSuffix(host, suffix) {
return true
}
}
return false
}
// --- Per-IP rate limiter (token bucket, used by /contribute) ---
type tokenBucket struct {
tokens float64
maxTokens float64
refillRate float64 // tokens per second
lastRefill time.Time
}
func (b *tokenBucket) allow() bool {
now := time.Now()
elapsed := now.Sub(b.lastRefill).Seconds()
b.tokens += elapsed * b.refillRate
if b.tokens > b.maxTokens {
b.tokens = b.maxTokens
}
b.lastRefill = now
if b.tokens >= 1 {
b.tokens--
return true
}
return false
}
type ipRateLimiter struct {
mu sync.Mutex
buckets map[string]*tokenBucket
rate float64
burst float64
}
func newIPRateLimiter(ratePerMin, burst float64) *ipRateLimiter {
rl := &ipRateLimiter{
buckets: make(map[string]*tokenBucket),
rate: ratePerMin / 60.0,
burst: burst,
}
go func() {
ticker := time.NewTicker(10 * time.Minute)
for range ticker.C {
rl.mu.Lock()
for ip, b := range rl.buckets {
if b.tokens >= b.maxTokens {
delete(rl.buckets, ip)
}
}
rl.mu.Unlock()
}
}()
return rl
}
func (rl *ipRateLimiter) allow(ip string) bool {
rl.mu.Lock()
defer rl.mu.Unlock()
b, ok := rl.buckets[ip]
if !ok {
b = &tokenBucket{
tokens: rl.burst,
maxTokens: rl.burst,
refillRate: rl.rate,
lastRefill: time.Now(),
}
rl.buckets[ip] = b
}
return b.allow()
}
var contributeRL = newIPRateLimiter(5, 5) // 5 requests/min, burst of 5
// --- Evalcenter helpers ---
func detectArch(rawURL string) string {
@ -745,7 +943,7 @@ func enableCORS(next http.Handler) http.Handler {
if origin == "https://msdl.tech-latest.com" || strings.HasPrefix(origin, "http://localhost:") {
w.Header().Set("Access-Control-Allow-Origin", origin)
}
w.Header().Set("Access-Control-Allow-Methods", "GET, OPTIONS")
w.Header().Set("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
w.Header().Set("Access-Control-Expose-Headers", "X-MSDL-Link-Status, X-MSDL-Link-Expires")
if r.Method == "OPTIONS" {
@ -888,6 +1086,29 @@ func handleProxy(w http.ResponseWriter, r *http.Request) {
return
}
// 1b. Redis L2 cache — check after in-memory miss, before calling Microsoft
if !forceRefresh && !hasCached && rdb != nil {
rctx, rcancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
redisRaw, redisErr := rdb.Get(rctx, redisKey(productID, skuID)).Bytes()
rcancel()
if redisErr == nil && len(redisRaw) > 0 {
exp := parseLinkExpiry(redisRaw)
if time.Now().Before(exp) {
linkCacheMu.Lock()
linkCache[cacheKey] = linkCacheEntry{RawJSON: redisRaw, ExpiresAt: exp, FetchedAt: time.Now()}
linkCacheMu.Unlock()
atomic.AddInt64(&mLinkCacheHits, 1)
log.Printf("/proxy: product_id=%s sku_id=%s -> redis L2 hit\n", productID, skuID)
w.Header().Set("X-MSDL-Link-Status", "cached")
if rawExp := extractRawExpiry(redisRaw); rawExp != "" {
w.Header().Set("X-MSDL-Link-Expires", rawExp)
}
w.Write(redisRaw)
return
}
}
}
// 2. Check negative cache — but only if no stale data and not a force refresh
if !forceRefresh && !hasCached {
negCacheMu.RLock()
@ -929,6 +1150,7 @@ func handleProxy(w http.ResponseWriter, r *http.Request) {
linkCacheMu.Lock()
linkCache[cacheKey] = linkCacheEntry{RawJSON: raw, ExpiresAt: exp, FetchedAt: time.Now()}
linkCacheMu.Unlock()
redisWriteLink(productID, skuID, raw, exp)
log.Printf("/proxy: background refresh succeeded for %s:%s, cached until %s\n", productID, skuID, exp.Format(time.RFC3339))
return nil, nil
})
@ -969,6 +1191,7 @@ func handleProxy(w http.ResponseWriter, r *http.Request) {
FetchedAt: time.Now(),
}
linkCacheMu.Unlock()
redisWriteLink(productID, skuID, raw, expiresAt)
atomic.AddInt64(&mLinkFetches, 1)
if forceRefresh {
@ -983,6 +1206,118 @@ func handleProxy(w http.ResponseWriter, r *http.Request) {
w.Write(raw)
}
// --- /contribute endpoint ---
func handleContribute(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
w.Header().Set("Content-Type", "application/json")
// Check CONTRIBUTE_SECRET header
if secret := os.Getenv("CONTRIBUTE_SECRET"); secret != "" {
if r.Header.Get("X-Contribute-Secret") != secret {
respondJSONError(w, http.StatusUnauthorized, "unauthorized")
return
}
}
// Per-IP rate limit (~5/min)
ip := r.RemoteAddr
if idx := strings.LastIndex(ip, ":"); idx != -1 {
ip = ip[:idx]
}
if fwd := r.Header.Get("X-Forwarded-For"); fwd != "" {
ip = strings.TrimSpace(strings.SplitN(fwd, ",", 2)[0])
}
if !contributeRL.allow(ip) {
respondJSONError(w, http.StatusTooManyRequests, "rate limit exceeded")
return
}
// Parse body
body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20)) // 1 MB cap
if err != nil {
respondJSONError(w, http.StatusBadRequest, "failed to read body")
return
}
var req struct {
ProductID string `json:"product_id"`
SkuID string `json:"sku_id"`
RawJSON json.RawMessage `json:"raw_json"`
}
if err := json.Unmarshal(body, &req); err != nil {
respondJSONError(w, http.StatusBadRequest, "invalid JSON")
return
}
productID := req.ProductID
skuID := req.SkuID
// Validation 1: product in catalog
if !validContributeProducts[productID] {
log.Printf("contribute: product_id=%s sku_id=%s -> rejected (unknown product)\n", productID, skuID)
respondJSONError(w, http.StatusUnprocessableEntity, "unknown product_id")
return
}
// Validation 2: sku_id format
if !validSkuID.MatchString(skuID) {
log.Printf("contribute: product_id=%s sku_id=%s -> rejected (invalid sku_id format)\n", productID, skuID)
respondJSONError(w, http.StatusUnprocessableEntity, "invalid sku_id")
return
}
raw := []byte(req.RawJSON)
if len(raw) == 0 {
respondJSONError(w, http.StatusBadRequest, "raw_json is required")
return
}
// Validation 3: expiry must be > 1 hour out
expiresAt := parseLinkExpiry(raw)
if time.Until(expiresAt) < time.Hour {
log.Printf("contribute: product_id=%s sku_id=%s -> rejected (expiry < 1h)\n", productID, skuID)
respondJSONError(w, http.StatusUnprocessableEntity, "link expires in less than 1 hour")
return
}
// Validation 4: all download URLs must be from allowed Microsoft CDN hosts
var data map[string]interface{}
if err := json.Unmarshal(raw, &data); err != nil {
respondJSONError(w, http.StatusUnprocessableEntity, "raw_json is not valid JSON")
return
}
opts, _ := data["ProductDownloadOptions"].([]interface{})
if len(opts) == 0 {
log.Printf("contribute: product_id=%s sku_id=%s -> rejected (no ProductDownloadOptions)\n", productID, skuID)
respondJSONError(w, http.StatusUnprocessableEntity, "raw_json has no ProductDownloadOptions")
return
}
for _, opt := range opts {
m, ok := opt.(map[string]interface{})
if !ok {
continue
}
uri, _ := m["Uri"].(string)
if uri == "" || !isAllowedCDNHost(uri) {
log.Printf("contribute: product_id=%s sku_id=%s -> rejected (disallowed host: %s)\n", productID, skuID, uri)
respondJSONError(w, http.StatusUnprocessableEntity, "download URL host not in allow-list")
return
}
}
// Write to in-memory cache + Redis
cacheKey := productID + ":" + skuID
linkCacheMu.Lock()
linkCache[cacheKey] = linkCacheEntry{RawJSON: raw, ExpiresAt: expiresAt, FetchedAt: time.Now()}
linkCacheMu.Unlock()
redisWriteLink(productID, skuID, raw, expiresAt)
log.Printf("contribute: product_id=%s sku_id=%s -> accepted, cached until %s\n", productID, skuID, expiresAt.Format(time.RFC3339))
w.WriteHeader(http.StatusNoContent)
}
// --- /evallinks endpoint ---
func handleEvalLinks(w http.ResponseWriter, r *http.Request) {
@ -1185,6 +1520,8 @@ func warmEvalCache() {
}
func main() {
initRedis()
go redisSeedLinkCache()
go cleanupSessions()
go cleanupCaches()
go warmEvalCache()
@ -1193,6 +1530,7 @@ func main() {
mux.HandleFunc("/skuinfo", handleSkuInfo)
mux.HandleFunc("/proxy", handleProxy)
mux.HandleFunc("/contribute", handleContribute)
mux.HandleFunc("/evallinks", handleEvalLinks)
mux.HandleFunc("/metrics", handleMetrics)
mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {

View file

@ -1,12 +1,49 @@
package main
import (
"bytes"
"context"
"encoding/json"
"flag"
"fmt"
"net/http"
"os"
"strings"
"sync"
"time"
)
// contributeURL is the MSDL backend endpoint that warms the shared link cache.
// Override with MSDL_API_URL env var, e.g. export MSDL_API_URL=http://localhost:3002
const contributeSecret = "msdl-contribute-2026"
func contributeURL() string {
if u := os.Getenv("MSDL_API_URL"); u != "" {
return strings.TrimRight(u, "/") + "/contribute"
}
return "https://api.msdl.tech-latest.com/contribute"
}
func postContribute(productID, skuID string, rawJSON []byte) {
payload := struct {
ProductID string `json:"product_id"`
SkuID string `json:"sku_id"`
RawJSON json.RawMessage `json:"raw_json"`
}{ProductID: productID, SkuID: skuID, RawJSON: rawJSON}
body, _ := json.Marshal(payload)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
req, _ := http.NewRequestWithContext(ctx, http.MethodPost, contributeURL(), bytes.NewReader(body))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-Contribute-Secret", contributeSecret)
resp, err := http.DefaultClient.Do(req)
if err != nil {
return
}
resp.Body.Close()
}
func main() {
if err := run(os.Args[1:]); err != nil {
fmt.Fprintln(os.Stderr, "error:", err)
@ -35,11 +72,13 @@ Flags:`)
langFlag := fs.String("lang", "", `language name, e.g. "English (United States)"`)
evalMode := fs.Bool("eval", false, "fetch evaluation ISOs")
listMode := fs.Bool("list", false, "list all products and exit")
noContributeFlag := fs.Bool("no-contribute", false, "skip sharing the link with the msdl.tech cache")
if err := fs.Parse(args); err != nil {
return err
}
query := strings.Join(fs.Args(), " ")
noContribute := *noContributeFlag || os.Getenv("MSDL_NO_CONTRIBUTE") == "1"
if *listMode {
fmt.Fprintln(os.Stderr, "Consumer products:")
@ -56,10 +95,10 @@ Flags:`)
if *evalMode {
return runEval(query)
}
return runConsumer(*productID, query, *langFlag)
return runConsumer(*productID, query, *langFlag, noContribute)
}
func runConsumer(productID, query, langName string) error {
func runConsumer(productID, query, langName string, noContribute bool) error {
var product Product
if productID != "" {
@ -115,14 +154,35 @@ func runConsumer(productID, query, langName string) error {
}
fmt.Fprintf(os.Stderr, "Fetching download link...\n")
links, err := fetchDownloadLinks(client, sessionID, product.ID, lang.ID)
links, rawJSON, err := fetchDownloadLinks(client, sessionID, product.ID, lang.ID)
if err != nil {
return fmt.Errorf("fetching download links: %w", err)
}
// Fire contribute in background before printing so the goroutine starts immediately.
var wg sync.WaitGroup
if !noContribute {
wg.Add(1)
go func() {
defer wg.Done()
postContribute(product.ID, lang.ID, rawJSON)
}()
}
for _, link := range links {
fmt.Println(link.URI)
}
if !noContribute {
fmt.Fprintln(os.Stderr, "Shared with msdl.tech cache to help other users (--no-contribute to opt out)")
// Wait up to 5s for the contribution to complete before process exit.
done := make(chan struct{})
go func() { wg.Wait(); close(done) }()
select {
case <-done:
case <-time.After(5 * time.Second):
}
}
return nil
}

View file

@ -266,7 +266,8 @@ func fetchLanguages(client *http.Client, sessionID, productID string) ([]Languag
return parseSkuInfo(raw)
}
func fetchDownloadLinks(client *http.Client, sessionID, productID, skuID string) ([]DownloadLink, error) {
// fetchDownloadLinks returns parsed links and the raw Microsoft JSON (for cache contribution).
func fetchDownloadLinks(client *http.Client, sessionID, productID, skuID string) ([]DownloadLink, []byte, error) {
wq := url.Values{}
wq.Set("profile", msProfile)
wq.Set("productEditionId", productID)
@ -289,9 +290,13 @@ func fetchDownloadLinks(client *http.Client, sessionID, productID, skuID string)
"https://www.microsoft.com/software-download-connector/api/GetProductDownloadLinksBySku?"+q.Encode(),
productID)
if err != nil {
return nil, err
return nil, nil, err
}
return parseDownloadLinks(raw)
links, err := parseDownloadLinks(raw)
if err != nil {
return nil, nil, err
}
return links, raw, nil
}
func extractFwlinks(pageHTML string) []string {