mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-27 19:37:00 +00:00
9575032b4c
* volume: forward fsync=true to replicas in ReplicatedWrite When a write request carries fsync=true, only the primary volume server flushed to disk: the replica fan-out URL in ReplicatedWrite only carried type/ttl/ts/cm, so replicas always wrote without fsync even when the client explicitly requested a durable write. Forward the fsync request parameter to the replica volume servers so a durable write means every replica has flushed to disk, not just the primary. Replicas without fsync are untouched (zero behavior change). * storage: flush a durable write inline while stopping The fsync flag on the write path really selects the async batch worker, and it was switched off once the store is stopping. So a fsync=true write landing during the pre-stop drain got acked without ever being flushed - and now that ReplicatedWrite forwards fsync, that covers replicas too. Flush it inline instead of queueing it. The drain keeps accepting writes, which is the whole point of preStopSeconds, and the ack still means the .dat is on disk. If the fsync fails, the append comes back off the .dat and the needle map goes back to what it pointed at before, so nothing resolves to an offset past the truncated end. * storage: make the store's stopping flag atomic SetStopping runs on the signal handler goroutine while the write and vacuum paths read the flag, so every read of it was racy. Nothing about the shutdown ordering changes; only the flag itself is now safe to read. * topology: check the errors the replication test was dropping The mock replica ignored its response write and the mock master ignored whatever Serve returned, so a broken mock would have shown up as a confusing timeout rather than a failure. Also drops the explicit listener close: grpc.Server.Stop already closes the listener it was given. --------- Co-authored-by: hzsunchao <hzsunchao@corp.netease.com> Co-authored-by: Chris Lu <chris.lu@gmail.com>
88 lines
3.1 KiB
Go
88 lines
3.1 KiB
Go
package storage
|
|
|
|
import (
|
|
"errors"
|
|
"os"
|
|
"testing"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
|
|
)
|
|
|
|
// A volume booted with .vif ReadOnly=true used to come back stuck: load()
|
|
// opens .idx as O_RDONLY and builds a SortedFileNeedleMap whose Put returns
|
|
// os.ErrInvalid, and MarkVolumeWritable only flipped the in-memory flag and
|
|
// rewrote .vif. Subsequent writes failed at v.nm.Put.
|
|
func TestMarkVolumeWritable_ReopensPersistedReadOnly(t *testing.T) {
|
|
dir := t.TempDir()
|
|
|
|
v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
|
|
if err != nil {
|
|
t.Fatalf("create volume: %v", err)
|
|
}
|
|
|
|
if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil {
|
|
t.Fatalf("initial write: %v", err)
|
|
}
|
|
|
|
// Persist read-only state into .vif, then simulate a server restart by
|
|
// closing and re-opening the volume from the same directory.
|
|
v.PersistReadOnly(true, false)
|
|
v.Close()
|
|
|
|
v2, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
|
|
if err != nil {
|
|
t.Fatalf("reload volume: %v", err)
|
|
}
|
|
defer v2.Close()
|
|
|
|
if !v2.noWriteOrDelete {
|
|
t.Fatalf("reloaded volume should be in noWriteOrDelete=true after .vif ReadOnly=true")
|
|
}
|
|
if _, ok := v2.nm.(*SortedFileNeedleMap); !ok {
|
|
t.Fatalf("reloaded readonly volume should use SortedFileNeedleMap, got %T", v2.nm)
|
|
}
|
|
|
|
// Pre-fix behaviour: SortedFileNeedleMap.Put returns os.ErrInvalid even
|
|
// once noWriteOrDelete is cleared. Confirm the failure mode the issue
|
|
// describes — flipping only the flag is not enough.
|
|
v2.noWriteOrDelete = false
|
|
_, _, _, writeErr := v2.writeNeedle2(newRandomNeedle(2), true, false, false)
|
|
if !errors.Is(writeErr, os.ErrInvalid) {
|
|
t.Fatalf("expected write through SortedFileNeedleMap to fail with os.ErrInvalid, got %v", writeErr)
|
|
}
|
|
v2.noWriteOrDelete = true // restore so reopenIdxForWrite reflects the real entry condition
|
|
|
|
if err := v2.reopenIdxForWrite(); err != nil {
|
|
t.Fatalf("reopenIdxForWrite: %v", err)
|
|
}
|
|
if _, stillSorted := v2.nm.(*SortedFileNeedleMap); stillSorted {
|
|
t.Fatalf("reopenIdxForWrite left SortedFileNeedleMap in place")
|
|
}
|
|
|
|
v2.noWriteOrDelete = false
|
|
if _, _, _, err := v2.writeNeedle2(newRandomNeedle(3), true, false, false); err != nil {
|
|
t.Fatalf("write after reopen: %v", err)
|
|
}
|
|
}
|
|
|
|
// reopenIdxForWrite must be a no-op when the volume already has a writable
|
|
// needle map — otherwise repeated MarkVolumeWritable calls would churn the
|
|
// index file handle for no reason.
|
|
func TestReopenIdxForWrite_NoopWhenAlreadyWritable(t *testing.T) {
|
|
dir := t.TempDir()
|
|
v, err := NewVolume(dir, dir, "", 2, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
|
|
if err != nil {
|
|
t.Fatalf("create volume: %v", err)
|
|
}
|
|
defer v.Close()
|
|
|
|
before := v.nm
|
|
if err := v.reopenIdxForWrite(); err != nil {
|
|
t.Fatalf("reopenIdxForWrite on writable volume: %v", err)
|
|
}
|
|
if v.nm != before {
|
|
t.Fatalf("reopenIdxForWrite replaced nm on a writable volume (before=%p after=%p)", before, v.nm)
|
|
}
|
|
}
|