v1.2.0: skip unchanged writes, transactional schema/root counters, snapshot replay for meta:events 0-0 (drops boot republish), collect-once search driver

internal/api/schema_handlers.go

@@ -5,38 +5,44 @@ import (
 	"net/http"
 )
 
-// handleSchemaGet returns the per-field schema (recomputed from the store every
-// few minutes; see schema.Indexer).
+// handleSchemaGet returns the per-field schema, folded from counters the write
+// path keeps exact (see schema.Indexer).
 func (s *Server) handleSchemaGet(w http.ResponseWriter, r *http.Request) {
 	if s.schemaIndexer == nil {
 		http.Error(w, "schema indexer not running", http.StatusServiceUnavailable)
 		return
 	}
-	snap := s.schemaIndexer.Snapshot()
+	snap, err := s.schemaIndexer.Snapshot(r.Context())
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
 	w.Header().Set("Content-Type", "application/json")
 	_ = json.NewEncoder(w).Encode(snap)
 }
 
-// handleSchemaRescan recomputes the schema from a full scan of the store and
-// returns it. It no longer republishes every property into meta:events — the
-// indexer reads the store itself — so a rescan floods no consumer.
+// handleSchemaRescan rebuilds the schema counters from a full read of the
+// records and returns the result. The counters are exact by construction, so
+// this is a repair tool, not routine: it runs as one write transaction, and
+// every other write waits for it (minutes on a large store).
 func (s *Server) handleSchemaRescan(w http.ResponseWriter, r *http.Request) {
 	if s.schemaIndexer == nil {
 		http.Error(w, "schema indexer not running", http.StatusServiceUnavailable)
 		return
 	}
-
-	snap, err := s.schemaIndexer.Rebuild(r.Context())
+	if err := s.storage.RebuildCounters(); err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	snap, err := s.schemaIndexer.Snapshot(r.Context())
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
-	out := *snap
-	out.Source = "rescan"
-
+	snap.Source = "rescan"
 	w.Header().Set("Content-Type", "application/json")
 	_ = json.NewEncoder(w).Encode(map[string]any{
 		"published_events": 0,
-		"schema":           &out,
+		"schema":           snap,
 	})
 }

internal/api/server.go

@@ -9,7 +9,6 @@ import (
 
 	"github.com/gorilla/mux"
 	"github.com/metazla/meta-core/internal/config"
-	"github.com/metazla/meta-core/internal/events"
 	"github.com/metazla/meta-core/internal/identity"
 	"github.com/metazla/meta-core/internal/leader"
 	"github.com/metazla/meta-core/internal/meshdisco"
@@ -36,7 +35,6 @@ type Server struct {
 	watchersManager   *watchers.Manager
 	watchersPoller    *watchers.Poller
 	watchersHandlers  *watchers.Handlers
-	metaPublisher     *events.MetaPublisher
 	schemaIndexer     *schema.Indexer
 	webdavHandler     *webdav.Handler
 	// challenges backs proof-of-possession on reveal/delete. In-process, so a
@@ -107,8 +105,7 @@ func NewServer(
 			// Create watchers poller
 			if s.fileWatcher != nil {
 				s.watchersPoller = watchers.NewPoller(watchersManager, s.fileWatcher, s.watcherDispatcher)
-				// Pass republish callback to handlers (handles nil metaPublisher case)
-				s.watchersHandlers = watchers.NewHandlers(watchersManager, s.watchersPoller, s.RepublishMetadata)
+				s.watchersHandlers = watchers.NewHandlers(watchersManager, s.watchersPoller)
 
 				// Hydrate the in-memory state registry from stored tuples so
 				// the first scan tick after boot can skip ComputeMidHash256
@@ -349,41 +346,22 @@ func (s *Server) Start() error {
 		s.mountStatsPoller.Start()
 	}
 
-	// Start meta publisher (if storage is connected). Live change events are
-	// written by the storage engine itself; the publisher owns the bulk
-	// republish below.
+	// Schema: served from counters the write path keeps exact; nothing runs.
+	// Change events need no publisher either — the storage engine appends them
+	// in the same transaction as each change, and a consumer that wants the
+	// whole state asks /api/events/meta for "0" and gets a snapshot (see
+	// streamSnapshot). The boot republish of every field is gone with it.
 	if s.storage != nil && s.storage.IsConnected() {
-		s.metaPublisher = events.NewMetaPublisher(s.storage)
-		if err := s.metaPublisher.Start(); err != nil {
-			log.Printf("[API] Warning: failed to start meta publisher: %v", err)
-		} else {
-			// Schema indexer: per-field schema (type/hint/breakdown) from a
-			// periodic scan of the store.
-			s.schemaIndexer = schema.NewIndexer(s.storage.Store())
-			if err := s.schemaIndexer.Start(); err != nil {
-				log.Printf("[API] Warning: failed to start schema indexer: %v", err)
-				s.schemaIndexer = nil
+		s.schemaIndexer = schema.NewIndexer(s.storage.Store())
+		go func() {
+			started := time.Now()
+			if n, err := s.storage.RebuildSearchIfEmpty(); err != nil {
+				log.Printf("[API] Warning: search row rebuild failed: %v", err)
+			} else if n > 0 {
+				log.Printf("[API] Rebuilt search rows for %d records in %s", n, time.Since(started).Round(time.Millisecond))
 			}
-
-			// Republish events for all existing metadata. Consumers that
-			// rebuild by replaying meta:events from the beginning (meta-fuse)
-			// depend on it.
-			go func() {
-				started := time.Now()
-				if n, err := s.storage.RebuildSearchIfEmpty(); err != nil {
-					log.Printf("[API] Warning: search row rebuild failed: %v", err)
-				} else if n > 0 {
-					log.Printf("[API] Rebuilt search rows for %d records in %s", n, time.Since(started).Round(time.Millisecond))
-				}
-
-				count, err := s.RepublishMetadata()
-				if err != nil {
-					log.Printf("[API] Warning: failed to republish metadata on startup: %v", err)
-				} else {
-					log.Printf("[API] Republished %d metadata events on startup", count)
-				}
-			}()
-		}
+		}()
+		go s.logWriteStats()
 	}
 
 	return nil
@@ -410,20 +388,6 @@ func (s *Server) Stop() error {
 		s.mountStatsPoller.Stop()
 	}
 
-	// Stop schema indexer before the publisher so the consumer drains cleanly
-	if s.schemaIndexer != nil {
-		if err := s.schemaIndexer.Stop(); err != nil {
-			log.Printf("[API] Warning: failed to stop schema indexer: %v", err)
-		}
-	}
-
-	// Stop meta publisher
-	if s.metaPublisher != nil {
-		if err := s.metaPublisher.Stop(); err != nil {
-			log.Printf("[API] Warning: failed to stop meta publisher: %v", err)
-		}
-	}
-
 	if s.server == nil {
 		return nil
 	}
@@ -436,26 +400,22 @@ func (s *Server) Stop() error {
 	return s.server.Shutdown(ctx)
 }
 
-// RepublishMetadata republishes all existing metadata to the meta:events stream
-// This is called on startup and after reset-all
-func (s *Server) RepublishMetadata() (int, error) {
-	if s.metaPublisher == nil {
-		return 0, fmt.Errorf("meta publisher not initialized")
-	}
-	if s.storage == nil || !s.storage.IsConnected() {
-		return 0, fmt.Errorf("storage not connected")
-	}
-
-	// Create the republish function that wraps storage methods
-	republishFunc := func() ([]string, func(string) (map[string]string, error), error) {
-		hashIDs, err := s.storage.GetAllHashIDs()
-		if err != nil {
-			return nil, nil, err
+// logWriteStats reports, every 15 minutes, how many writes changed a value
+// and how many found it already in place (kv skips those). The ratio is what
+// the no-op skip buys on a permanent store that re-persists repeated hits.
+func (s *Server) logWriteStats() {
+	ticker := time.NewTicker(15 * time.Minute)
+	defer ticker.Stop()
+	for range ticker.C {
+		store := s.storage.Store()
+		if store == nil {
+			return
+		}
+		changed, unchanged := store.WriteStats()
+		if total := changed + unchanged; total > 0 {
+			log.Printf("[Storage] writes since start: %d changed, %d unchanged skipped (%.0f%% no-op)", changed, unchanged, 100*float64(unchanged)/float64(total))
 		}
-		return hashIDs, s.storage.GetMetadataFlat, nil
 	}
-
-	return s.metaPublisher.RepublishAllMetadata(republishFunc)
 }
 
 // statusRecorder captures the response status so loggingMiddleware can stay

internal/api/sse_events.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/metazla/meta-core/internal/kv"
+	"github.com/metazla/meta-core/internal/storage"
 )
 
 // SSE event stream endpoints for the file:events and meta:events streams.
@@ -97,7 +98,22 @@ func (s *Server) streamSSE(w http.ResponseWriter, r *http.Request, stream string
 	}
 
 	ctx := r.Context()
-	cursor := resolveStartCursor(ctx, store, stream, r.Header.Get("Last-Event-ID"), w, flusher)
+	lastID := r.Header.Get("Last-Event-ID")
+	var cursor string
+	if stream == storage.MetaEventsStream && (lastID == "0" || lastID == "0-0") {
+		// "From the beginning" on meta:events means "the whole current
+		// state": stream it from the records themselves, then the live tail.
+		head, err := streamSnapshot(ctx, store, w, flusher)
+		if err != nil {
+			if ctx.Err() == nil {
+				log.Printf("[SSE %s] snapshot failed: %v", stream, err)
+			}
+			return
+		}
+		cursor = head
+	} else {
+		cursor = resolveStartCursor(ctx, store, stream, lastID, w, flusher)
+	}
 
 	lastWrite := time.Now()
 	for {
@@ -147,6 +163,54 @@ func (s *Server) streamSSE(w http.ResponseWriter, r *http.Request, stream string
 	}
 }
 
+// streamSnapshot sends one `set` event per existing record key, then returns
+// the stream position the live tail continues from.
+//
+// This is how a consumer that rebuilds its state from events (meta-fuse sends
+// Last-Event-ID 0-0 on every start) gets ALL of it. The stream itself only
+// retains ~100k entries, so replaying it never could; meta-core used to
+// republish every field into it on boot instead, which took minutes and still
+// lost everything past the retention window.
+//
+// The head is captured BEFORE the snapshot is read, so a change racing the
+// snapshot appears in the tail too — re-applying a `set` is idempotent. Only
+// the final snapshot event carries an `id:`, so a consumer that disconnects
+// mid-snapshot still holds its old cursor and gets a fresh snapshot.
+func streamSnapshot(ctx context.Context, store *kv.Store, w http.ResponseWriter, flusher http.Flusher) (string, error) {
+	head, err := store.LastID(ctx, storage.MetaEventsStream)
+	if err != nil {
+		return "", err
+	}
+	ts := strconv.FormatInt(time.Now().UnixMilli(), 10)
+	pending := ""
+	err = store.ScanPrefix(ctx, "file:", 5000, func(rows []kv.KV) error {
+		for _, row := range rows {
+			if pending != "" {
+				writeSnapshotEvent(w, pending, ts, "")
+			}
+			pending = row.K
+		}
+		flusher.Flush()
+		return ctx.Err()
+	})
+	if err != nil {
+		return "", err
+	}
+	if pending != "" {
+		writeSnapshotEvent(w, pending, ts, head)
+		flusher.Flush()
+	}
+	return head, nil
+}
+
+func writeSnapshotEvent(w http.ResponseWriter, key, ts, id string) {
+	payload, _ := json.Marshal(map[string]string{"key": key, "ts": ts})
+	if id != "" {
+		fmt.Fprintf(w, "id: %s\n", id)
+	}
+	fmt.Fprintf(w, "event: set\ndata: %s\n\n", payload)
+}
+
 // resolveStartCursor decides where to start reading.
 //   - No Last-Event-ID → the stream's last issued ID, resolved ONCE here (only
 //     entries appended from now on). Redis' "$" was re-resolved on every

internal/api/sse_snapshot_test.go

@@ -0,0 +1,89 @@
+package api
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+// syncRecorder is a ResponseRecorder safe to read while the handler writes.
+type syncRecorder struct {
+	*httptest.ResponseRecorder
+	mu sync.Mutex
+}
+
+func (r *syncRecorder) Write(b []byte) (int, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.ResponseRecorder.Write(b)
+}
+
+func (r *syncRecorder) body() string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.Body.String()
+}
+
+// Last-Event-ID 0-0 on meta:events returns the whole current state — one set
+// per record key, only the last carrying an id — then the live tail.
+func TestMetaEventsSnapshotThenTail(t *testing.T) {
+	stor := newTestStorage(t)
+	s := &Server{storage: stor}
+	for i, id := range []string{"R1", "R2"} {
+		if err := stor.SetMetadataFlat(id, map[string]string{"title": "T", "n": string(rune('a' + i))}); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	req := httptest.NewRequest(http.MethodGet, "/api/events/meta", nil).WithContext(ctx)
+	req.Header.Set("Last-Event-ID", "0-0")
+	rec := &syncRecorder{ResponseRecorder: httptest.NewRecorder()}
+	done := make(chan struct{})
+	go func() { s.handleEventsMeta(rec, req); close(done) }()
+
+	waitFor := func(substr string) {
+		t.Helper()
+		deadline := time.Now().Add(5 * time.Second)
+		for !strings.Contains(rec.body(), substr) {
+			if time.Now().After(deadline) {
+				t.Fatalf("timed out waiting for %q in:\n%s", substr, rec.body())
+			}
+			time.Sleep(20 * time.Millisecond)
+		}
+	}
+	waitFor(`"key":"file:R2/title"`)
+	if err := stor.SetProperty("R1", "title", "changed"); err != nil {
+		t.Fatal(err)
+	}
+	waitFor(`event: set`)
+	time.Sleep(300 * time.Millisecond)
+	cancel()
+	<-done
+
+	body := rec.body()
+	for _, k := range []string{"file:R1/title", "file:R1/n", "file:R2/title", "file:R2/n"} {
+		if !strings.Contains(body, `"key":"`+k+`"`) {
+			t.Errorf("snapshot is missing %s", k)
+		}
+	}
+	events := strings.Split(strings.TrimSpace(body), "\n\n")
+	if len(events) != 5 {
+		t.Fatalf("want 4 snapshot events + 1 live event, got %d:\n%s", len(events), body)
+	}
+	for i, e := range events[:3] {
+		if strings.Contains(e, "id: ") {
+			t.Errorf("snapshot event %d carries an id: %q", i, e)
+		}
+	}
+	if !strings.HasPrefix(events[3], "id: ") || !strings.HasPrefix(events[4], "id: ") {
+		t.Fatalf("the last snapshot event and the live event must carry ids:\n%s", body)
+	}
+	if !strings.Contains(events[4], `"key":"file:R1/title"`) {
+		t.Fatalf("live tail did not follow the snapshot: %q", events[4])
+	}
+}