RC-5

internal/server/guess_lottery.go

@@ -0,0 +1,135 @@
+package server
+
+import (
+	"context"
+	crand "crypto/rand"
+	"errors"
+	"math/big"
+	"sync"
+	"time"
+)
+
+var (
+	errLotteryDuplicate = errors.New("guess already entered in current lottery window")
+	errLotteryFull      = errors.New("guess lottery window is full")
+)
+
+const maxLotteryTicketsPerWindow = 250000
+
+type lotteryTicket struct {
+	key    string
+	ctx    context.Context
+	result chan bool
+}
+
+type lotteryBucket struct {
+	max     int
+	end     time.Time
+	tickets []*lotteryTicket
+	keys    map[string]struct{}
+}
+
+type guessLottery struct {
+	mu      sync.Mutex
+	buckets map[string]*lotteryBucket
+}
+
+func newGuessLottery() *guessLottery {
+	return &guessLottery{buckets: make(map[string]*lotteryBucket)}
+}
+
+// enter batches all valid tips for a task into aligned time windows. At the
+// window boundary exactly up to max tickets are selected uniformly at random.
+// The request intentionally waits for the draw so later arrivals in the same
+// window have the same chance as earlier arrivals.
+func (l *guessLottery) enter(ctx context.Context, taskID, clientID string, seq int64, window time.Duration, max int) (bool, error) {
+	if max <= 0 || window <= 0 {
+		return true, nil
+	}
+	now := time.Now().UTC()
+	windowNS := window.Nanoseconds()
+	if windowNS <= 0 {
+		return true, nil
+	}
+	idx := now.UnixNano() / windowNS
+	end := time.Unix(0, (idx+1)*windowNS).UTC()
+	bucketKey := taskID + "|" + end.Format(time.RFC3339Nano) + "|" + window.String()
+	ticketKey := clientID + "|" + big.NewInt(seq).String()
+	t := &lotteryTicket{key: ticketKey, ctx: ctx, result: make(chan bool, 1)}
+
+	l.mu.Lock()
+	b := l.buckets[bucketKey]
+	if b == nil {
+		b = &lotteryBucket{max: max, end: end, keys: make(map[string]struct{})}
+		l.buckets[bucketKey] = b
+		delay := time.Until(end)
+		if delay < 0 {
+			delay = 0
+		}
+		time.AfterFunc(delay, func() { l.draw(bucketKey) })
+	}
+	if _, exists := b.keys[ticketKey]; exists {
+		l.mu.Unlock()
+		return false, errLotteryDuplicate
+	}
+	if len(b.tickets) >= maxLotteryTicketsPerWindow {
+		l.mu.Unlock()
+		return false, errLotteryFull
+	}
+	b.keys[ticketKey] = struct{}{}
+	b.tickets = append(b.tickets, t)
+	l.mu.Unlock()
+
+	select {
+	case selected := <-t.result:
+		return selected, nil
+	case <-ctx.Done():
+		return false, ctx.Err()
+	}
+}
+
+func (l *guessLottery) draw(bucketKey string) {
+	l.mu.Lock()
+	b := l.buckets[bucketKey]
+	if b == nil {
+		l.mu.Unlock()
+		return
+	}
+	delete(l.buckets, bucketKey)
+	tickets := append([]*lotteryTicket(nil), b.tickets...)
+	max := b.max
+	l.mu.Unlock()
+
+	// Canceled HTTP requests do not consume one of the scarce winning slots.
+	alive := tickets[:0]
+	for _, t := range tickets {
+		select {
+		case <-t.ctx.Done():
+			// skip
+		default:
+			alive = append(alive, t)
+		}
+	}
+	tickets = alive
+	if max > len(tickets) {
+		max = len(tickets)
+	}
+	// Partial Fisher-Yates with crypto/rand gives every ticket equal odds.
+	for i := 0; i < max; i++ {
+		nBig, err := crand.Int(crand.Reader, big.NewInt(int64(len(tickets)-i)))
+		if err != nil {
+			// crypto/rand failure is extremely unusual; deterministic fallback still
+			// keeps the quota safe, but does not claim cryptographic randomness.
+			nBig = big.NewInt(0)
+		}
+		j := i + int(nBig.Int64())
+		tickets[i], tickets[j] = tickets[j], tickets[i]
+	}
+	for i, t := range tickets {
+		selected := i < max
+		select {
+		case t.result <- selected:
+		default:
+		}
+	}
+}

internal/server/guess_lottery_test.go

@@ -0,0 +1,44 @@
+package server
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+)
+
+func TestGuessLotteryDrawSelectsExactQuota(t *testing.T) {
+	l := newGuessLottery()
+	const total = 40
+	const quota = 9
+	b := &lotteryBucket{max: quota, end: time.Now().Add(time.Second), keys: make(map[string]struct{})}
+	for i := 0; i < total; i++ {
+		ticket := &lotteryTicket{key: fmt.Sprintf("c-%d|0", i), ctx: context.Background(), result: make(chan bool, 1)}
+		b.tickets = append(b.tickets, ticket)
+		b.keys[ticket.key] = struct{}{}
+	}
+	l.buckets["test"] = b
+	l.draw("test")
+	selected := 0
+	for _, ticket := range b.tickets {
+		if <-ticket.result {
+			selected++
+		}
+	}
+	if selected != quota {
+		t.Fatalf("selected %d tickets, want %d", selected, quota)
+	}
+}
+
+func TestGuessLotteryCanceledTicketDoesNotConsumeQuota(t *testing.T) {
+	l := newGuessLottery()
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	canceled := &lotteryTicket{key: "canceled", ctx: ctx, result: make(chan bool, 1)}
+	alive := &lotteryTicket{key: "alive", ctx: context.Background(), result: make(chan bool, 1)}
+	l.buckets["test"] = &lotteryBucket{max: 1, tickets: []*lotteryTicket{canceled, alive}, keys: map[string]struct{}{"canceled": {}, "alive": {}}}
+	l.draw("test")
+	if got := <-alive.result; !got {
+		t.Fatal("live ticket should receive the available slot")
+	}
+}

internal/server/profile_cleanup_test.go

@@ -0,0 +1,55 @@
+package server
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"neuralhunt/internal/auth"
+	"neuralhunt/internal/data"
+	rtx "neuralhunt/internal/runtime"
+)
+
+func TestProfileCleanupPreviewProtectsConnectedClient(t *testing.T) {
+	ctx := context.Background()
+	db, err := data.OpenSQLite(ctx, t.TempDir()+"/cleanup.db")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	store := data.New(db)
+	jwk := auth.PublicJWK{Kty: "EC", Crv: "P-256", X: "AQ", Y: "Ag"}
+	for _, id := range []string{"offline_old", "online_old"} {
+		if err := store.UpsertClient(ctx, id, jwk); err != nil {
+			t.Fatal(err)
+		}
+	}
+	old := time.Now().UTC().Add(-10 * 24 * time.Hour).UnixMilli()
+	if _, err := db.ExecContext(ctx, `UPDATE clients SET last_seen=?`, old); err != nil {
+		t.Fatal(err)
+	}
+	runtimeState := rtx.New()
+	if _, err := runtimeState.AcquirePresence("online_old", "session"); err != nil {
+		t.Fatal(err)
+	}
+	s := &Server{store: store, runtime: runtimeState}
+	preview, ids, err := s.profileCleanupPreview(ctx, 7*24*60*60)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if preview.Eligible != 1 || preview.ProtectedConnected != 1 {
+		t.Fatalf("unexpected preview: %+v", preview)
+	}
+	if len(ids) != 1 || ids[0] != "offline_old" {
+		t.Fatalf("unexpected ids: %#v", ids)
+	}
+}
+
+func TestProfileCleanupDurationGuard(t *testing.T) {
+	if _, err := profileCleanupDurationSeconds("3599"); err == nil {
+		t.Fatal("sub-hour cleanup window should be rejected")
+	}
+	if got, err := profileCleanupDurationSeconds("86400"); err != nil || got != 86400 {
+		t.Fatalf("got=%d err=%v", got, err)
+	}
+}

internal/server/server.go

@@ -37,6 +37,7 @@ type Server struct {
 	hub                                          *wsx.Hub
 	runtime                                      *rtx.State
 	artifactWorker                               *artifact.Worker
+	lottery                                      *guessLottery
 	adminUser, adminPass, staticDir, artifactDir string
 	upgrader                                     websocket.Upgrader
 }
@@ -49,6 +50,7 @@ func New(store *data.Store, a *auth.Manager, sm *settings.Manager, hub *wsx.Hub,
 		hub:            hub,
 		runtime:        runtimeState,
 		artifactWorker: artifactWorker,
+		lottery:        newGuessLottery(),
 		adminUser:      env("ADMIN_USER", "admin"),
 		adminPass:      env("ADMIN_PASSWORD", "change-me"),
 		staticDir:      env("STATIC_DIR", ""),
@@ -178,6 +180,8 @@ func (s *Server) Routes() http.Handler {
 		r.Use(func(n http.Handler) http.Handler { return s.require("admin", n) })
 		r.Get("/api/admin/overview", s.adminOverview)
 		r.Get("/api/admin/performance", s.adminPerformance)
+		r.Get("/api/admin/profiles/cleanup-preview", s.adminProfileCleanupPreview)
+		r.Post("/api/admin/profiles/cleanup", s.adminProfileCleanup)
 		r.Get("/api/admin/settings", s.adminSettingsGet)
 		r.Put("/api/admin/settings", s.adminSettingsPut)
 		r.Get("/api/admin/tasks", s.adminTasks)
@@ -349,6 +353,8 @@ func taskDTO(t data.Task, next int64, sm settings.Runtime) map[string]any {
 		"next_seq":                   next,
 		"server_min_interval_sec":    serverMin,
 		"client_submit_interval_sec": clientSubmit,
+		"guess_lottery_window_sec":   sm.GuessLotteryWindowSec,
+		"guess_lottery_max_accepted": sm.GuessLotteryMaxAccepted,
 		"default_max_nodes":          sm.DefaultMaxNodes,
 		"paused":                     t.Paused,
 		"revision":                   t.Revision,
@@ -495,6 +501,78 @@ func (s *Server) guess(w http.ResponseWriter, r *http.Request) {
 		jsonOut(w, 401, false)
 		return
 	}
+	if _, ok := s.runtime.Current(t.Task, c.ClientID); !ok {
+		snap, _ := s.store.LoadGuessState(r.Context(), t.ID, c.ClientID)
+		s.runtime.InitGuess(t.Task, c.ClientID, rtx.GuessState{NextSeq: snap.NextSeq, LastGuess: snap.LastGuess, BestScore: snap.BestScore, GuessCount: snap.GuessCount})
+	}
+	cfg := s.settings.Get()
+	serverMin, _ := taskIntervals(t.Task, cfg)
+	minInterval := time.Duration(serverMin) * time.Second
+	if err := s.runtime.CanSubmit(t.Task, c.ClientID, in.Seq, minInterval); err != nil {
+		switch {
+		case errors.Is(err, rtx.ErrRateLimited):
+			jsonOut(w, 429, false)
+		case errors.Is(err, rtx.ErrBadSequence):
+			expected := int64(0)
+			if cur, ok := s.runtime.Current(t.Task, c.ClientID); ok {
+				expected = cur.NextSeq
+			}
+			jsonAPIError(w, http.StatusConflict, "sequence_mismatch", "guess sequence is stale", map[string]any{"next_seq": expected})
+		default:
+			jsonOut(w, 500, false)
+		}
+		return
+	}
+
+	if cfg.GuessLotteryMaxAccepted > 0 {
+		selected, drawErr := s.lottery.enter(r.Context(), t.ID, c.ClientID, in.Seq, time.Duration(cfg.GuessLotteryWindowSec)*time.Second, cfg.GuessLotteryMaxAccepted)
+		if drawErr != nil {
+			switch {
+			case errors.Is(drawErr, errLotteryDuplicate):
+				jsonAPIError(w, http.StatusConflict, "lottery_duplicate", "guess is already waiting for the current draw", nil)
+			case errors.Is(drawErr, errLotteryFull):
+				jsonAPIError(w, http.StatusTooManyRequests, "lottery_full", "guess lottery window is full", nil)
+			case errors.Is(drawErr, context.Canceled), errors.Is(drawErr, context.DeadlineExceeded):
+				return
+			default:
+				jsonOut(w, 500, false)
+			}
+			return
+		}
+
+		// A task may have been rerolled, paused or completed while this request
+		// waited for the draw. Never consume/evaluate a stale lottery ticket.
+		fresh, freshErr := s.store.SecretTask(r.Context(), id)
+		if freshErr != nil || fresh.Status != "active" {
+			jsonAPIError(w, http.StatusConflict, "task_inactive", "task is no longer active", nil)
+			return
+		}
+		if fresh.Paused {
+			jsonOut(w, 423, false)
+			return
+		}
+		if fresh.PublicSeed != t.PublicSeed || fresh.Revision != t.Revision {
+			jsonAPIError(w, http.StatusConflict, "task_config_changed", "task configuration changed during lottery; resync required", map[string]any{"revision": fresh.Revision})
+			return
+		}
+		t = fresh
+		if !selected {
+			next, skipErr := s.runtime.SkipLottery(t.Task, c.ClientID, in.Seq, minInterval)
+			if skipErr != nil {
+				if errors.Is(skipErr, rtx.ErrBadSequence) {
+					jsonAPIError(w, http.StatusConflict, "sequence_mismatch", "guess sequence changed while waiting for lottery", map[string]any{"next_seq": next})
+				} else if errors.Is(skipErr, rtx.ErrRateLimited) {
+					jsonOut(w, 429, false)
+				} else {
+					jsonOut(w, 500, false)
+				}
+				return
+			}
+			jsonAPIError(w, http.StatusTooManyRequests, "lottery_not_selected", "guess was not selected in this lottery window", map[string]any{"next_seq": next})
+			return
+		}
+	}
+
 	d, err := core.Distance(in.Guess, t.Secret)
 	if err != nil {
 		jsonOut(w, 400, false)
@@ -502,12 +580,7 @@ func (s *Server) guess(w http.ResponseWriter, r *http.Request) {
 	}
 	correct := d.Sign() == 0
 	score := core.Score(d, t.RangeBits)
-	if _, ok := s.runtime.Current(t.Task, c.ClientID); !ok {
-		snap, _ := s.store.LoadGuessState(r.Context(), t.ID, c.ClientID)
-		s.runtime.InitGuess(t.Task, c.ClientID, rtx.GuessState{NextSeq: snap.NextSeq, LastGuess: snap.LastGuess, BestScore: snap.BestScore, GuessCount: snap.GuessCount})
-	}
-	serverMin, _ := taskIntervals(t.Task, s.settings.Get())
-	accepted, err := s.runtime.Accept(t.Task, c.ClientID, in.Seq, score, time.Duration(serverMin)*time.Second)
+	accepted, err := s.runtime.Accept(t.Task, c.ClientID, in.Seq, score, minInterval)
 	if err != nil {
 		switch {
 		case errors.Is(err, rtx.ErrRateLimited):
@@ -736,6 +809,108 @@ func (s *Server) serveAdminArtifactPart(w http.ResponseWriter, r *http.Request,
 	http.ServeFile(w, r, path)
 }
 
+type profileCleanupPreview struct {
+	CutoffMS           int64 `json:"cutoff_ms"`
+	InactiveForSeconds int64 `json:"inactive_for_seconds"`
+	Eligible           int   `json:"eligible"`
+	ProtectedWinners   int64 `json:"protected_winners"`
+	ProtectedConnected int   `json:"protected_connected"`
+	OldestEligibleMS   int64 `json:"oldest_eligible_ms,omitempty"`
+	NewestEligibleMS   int64 `json:"newest_eligible_ms,omitempty"`
+}
+
+func profileCleanupDurationSeconds(raw string) (int64, error) {
+	seconds, err := strconv.ParseInt(strings.TrimSpace(raw), 10, 64)
+	if err != nil {
+		return 0, errors.New("inactive_for_seconds must be an integer")
+	}
+	// A one-hour minimum prevents an accidental near-live purge while still
+	// allowing short-lived development/test deployments to clean up quickly.
+	if seconds < 3600 || seconds > 10*365*24*60*60 {
+		return 0, errors.New("inactive_for_seconds must be between 3600 seconds and 10 years")
+	}
+	return seconds, nil
+}
+
+func (s *Server) profileCleanupPreview(ctx context.Context, inactiveForSeconds int64) (profileCleanupPreview, []string, error) {
+	cutoff := time.Now().UTC().Add(-time.Duration(inactiveForSeconds) * time.Second).UnixMilli()
+	candidates, err := s.store.InactiveNonWinnerClients(ctx, cutoff)
+	if err != nil {
+		return profileCleanupPreview{}, nil, err
+	}
+	protectedWinners, err := s.store.OldWinnerCount(ctx, cutoff)
+	if err != nil {
+		return profileCleanupPreview{}, nil, err
+	}
+	ids := make([]string, 0, len(candidates))
+	out := profileCleanupPreview{CutoffMS: cutoff, InactiveForSeconds: inactiveForSeconds, ProtectedWinners: protectedWinners}
+	for _, c := range candidates {
+		if s.runtime.IsConnected(c.ClientID) {
+			out.ProtectedConnected++
+			continue
+		}
+		ids = append(ids, c.ClientID)
+		if out.OldestEligibleMS == 0 || c.LastSeen < out.OldestEligibleMS {
+			out.OldestEligibleMS = c.LastSeen
+		}
+		if c.LastSeen > out.NewestEligibleMS {
+			out.NewestEligibleMS = c.LastSeen
+		}
+	}
+	out.Eligible = len(ids)
+	return out, ids, nil
+}
+
+func (s *Server) adminProfileCleanupPreview(w http.ResponseWriter, r *http.Request) {
+	seconds, err := profileCleanupDurationSeconds(r.URL.Query().Get("inactive_for_seconds"))
+	if err != nil {
+		jsonOut(w, 400, map[string]string{"error": err.Error()})
+		return
+	}
+	preview, _, err := s.profileCleanupPreview(r.Context(), seconds)
+	if err != nil {
+		jsonOut(w, 500, map[string]string{"error": "profile cleanup preview failed: " + err.Error()})
+		return
+	}
+	jsonOut(w, 200, preview)
+}
+
+func (s *Server) adminProfileCleanup(w http.ResponseWriter, r *http.Request) {
+	var in struct {
+		InactiveForSeconds int64 `json:"inactive_for_seconds"`
+	}
+	if err := decode(r, &in); err != nil {
+		jsonOut(w, 400, map[string]string{"error": "bad json: " + err.Error()})
+		return
+	}
+	seconds, err := profileCleanupDurationSeconds(strconv.FormatInt(in.InactiveForSeconds, 10))
+	if err != nil {
+		jsonOut(w, 400, map[string]string{"error": err.Error()})
+		return
+	}
+	preview, ids, err := s.profileCleanupPreview(r.Context(), seconds)
+	if err != nil {
+		jsonOut(w, 500, map[string]string{"error": "profile cleanup check failed: " + err.Error()})
+		return
+	}
+	deleted, err := s.store.DeleteInactiveNonWinnerClients(r.Context(), preview.CutoffMS, ids)
+	if err != nil {
+		jsonOut(w, 500, map[string]string{"error": "profile cleanup failed: " + err.Error()})
+		return
+	}
+	for _, id := range deleted {
+		s.runtime.ForgetClient(id)
+	}
+	jsonOut(w, 200, map[string]any{
+		"deleted":              len(deleted),
+		"eligible_before":      preview.Eligible,
+		"protected_winners":    preview.ProtectedWinners,
+		"protected_connected":  preview.ProtectedConnected,
+		"cutoff_ms":            preview.CutoffMS,
+		"inactive_for_seconds": seconds,
+	})
+}
+
 func (s *Server) adminOverview(w http.ResponseWriter, r *http.Request) {
 	var clients, activeTasks, completedTasks, guesses, artifacts int64
 	_ = s.store.DB.QueryRowContext(r.Context(), `SELECT count(*) FROM clients`).Scan(&clients)
@@ -1062,6 +1237,10 @@ func (s *Server) ws(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "identity not registered", http.StatusUnauthorized)
 		return
 	}
+	// WebSocket use counts as recent profile activity. This also makes a race
+	// with the admin cleanup safe: a newly connecting identity no longer
+	// matches an old last_seen cutoff.
+	s.store.TouchClient(r.Context(), c.ClientID)
 	t, err := s.store.TaskForClient(r.Context(), c.ClientID)
 	if err != nil {
 		http.Error(w, "no task", 503)
@@ -1080,7 +1259,13 @@ func (s *Server) ws(w http.ResponseWriter, r *http.Request) {
 	}
 	cl := wsx.NewClient(conn, t.ID, c.ClientID, false)
 	s.hub.Add(cl)
-	defer func() { s.hub.Remove(cl); s.runtime.ReleasePresence(c.ClientID, c.SessionID, leaseID) }()
+	defer func() {
+		s.hub.Remove(cl)
+		s.runtime.ReleasePresence(c.ClientID, c.SessionID, leaseID)
+		// Mark the disconnect time as last activity. A client that stayed online
+		// for days therefore starts its inactivity window only after disconnect.
+		s.store.TouchClient(context.Background(), c.ClientID)
+	}()
 
 	// A map point is durable only once per client/task. Subsequent losing guesses
 	// stay in memory; improvements are checkpointed by the guess handler.