tailer,store: fail loudly when the watermark cannot be read

A failed Watermark read (IO error, corruption, closed store) defaulted
the watermark to 0, silently re-draining the entire log from index 0.
Skip the log instead and log the failure.

Also make getRaw hold the store mutex and return errStoreClosed after
Close: pebble panics on Get-after-Close, so a tailer (or any reader)
racing shutdown crashed the process instead of receiving an error.

Test: TestWatermarkErrorSkipsLog proves the tailer never contacts the
log when the watermark read fails.

internal/store/store.go

@@ -466,6 +466,15 @@ func (s *Store) recountInLoop(flush func() error, batch *pebble.Batch) (uint64,
 }
 
 func (s *Store) getRaw(k []byte) ([]byte, error) {
+	// Hold mu across the read: Close marks the store closed under mu before
+	// closing pebble, so a read that starts before Close finishes safely,
+	// and one that starts after returns an error instead of panicking
+	// inside pebble ("pebble: closed").
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.closed {
+		return nil, errStoreClosed
+	}
 	v, closer, err := s.db.Get(k)
 	if err == pebble.ErrNotFound {
 		return nil, nil

internal/tailer/tailer.go

@@ -69,7 +69,11 @@ func (t *Tailer) tailOne(ctx context.Context, lg loglist.Log, state string) {
 	id := lg.LogID
 	wm, err := t.Store.Watermark(id)
 	if err != nil {
-		wm = 0
+		// A read failure is indistinguishable from corruption. Assuming
+		// zero here would re-drain the entire log from index 0, so skip
+		// the log instead; the watermark is retried on next process start.
+		log.Printf("tail %s: watermark read failed: %v; skipping log to avoid full re-drain", client.ShortID(), err)
+		return
 	}
 	drainTarget := int64(-1)
 	switch state {

internal/tailer/tailer_test.go

@@ -232,3 +232,35 @@ func TestSTHRegressionIgnored(t *testing.T) {
 		t.Errorf("watermark rewound to %d", wm)
 	}
 }
+
+func TestWatermarkErrorSkipsLog(t *testing.T) {
+	fl := newFakeLog(t)
+	keyB64, logID := fl.keyB64(t)
+	der := makeLeafCert(t, []string{"wmfail.example.com"})
+	fl.entries = [][]byte{fl.leafInput(der, 1000)}
+	srv := fl.serve(t)
+	defer srv.Close()
+
+	st, err := store.Open(filepath.Join(t.TempDir(), "db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	// A closed store makes Watermark fail, as it would on IO trouble or
+	// corruption. The tailer must skip the log, not assume watermark 0
+	// and re-drain it from the beginning.
+	if err := st.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	tr := &Tailer{Store: st, Interval: time.Millisecond}
+	tr.Sync(context.Background(), []loglist.Log{{
+		LogID: logID,
+		Key:   keyB64,
+		URL:   srv.URL,
+		State: map[string]loglist.StateDetail{"usable": {}},
+	}})
+	tr.Wait()
+	if fl.sthHits != 0 {
+		t.Fatalf("tailer contacted log %d times despite watermark read failure", fl.sthHits)
+	}
+}