From 673bc773a5aa05cf648721711b70a2b051a2964c Mon Sep 17 00:00:00 2001 From: Chirag Jaiswal Date: Thu, 6 Aug 2026 08:40:57 +0530 Subject: [PATCH] Make the master's namespace durable and stop silent data loss Live testing of the cluster surfaced four failures that all reported success while losing or corrupting data. Each is fixed here and pinned by a test that fails against the old behaviour. Master restart orphaned every file. Metadata lived only in Go maps, so a restart left chunks intact on disk but permanently unreachable, and the orphans were never reclaimed. The master now follows the GFS operation log design (HDFS edits + fsimage): the namespace is appended to a write-ahead log and fsync'd before the RPC is acknowledged, with checkpoints published by atomic rename to bound replay. Replica locations are deliberately not persisted -- chunkservers are the only authority on what they hold -- so they are relearned from registration and heartbeats, and orphan collection waits out a grace period first. Downloading a file twice into one directory appended to the previous result, doubling it. Downloads now stream into a temp file and rename into place, so a re-run replaces the old copy and an interrupted run leaves nothing behind. A download that could reach no replica returned nil and exited zero with no file written. Both the download and upload paths now propagate errors; the chunker distinguishes io.EOF from a real read failure rather than treating every error as end-of-file. Uploads wrote a single replica while the master recorded the full replication factor, leaving a window at RF=1 that the master could not see. The client now writes every allocated replica before reporting success. Also fixed, from the same review: - The master held its global mutex across the blocking stream.Send in Heartbeat, so one slow chunkserver could stall every RPC in the cluster. The response is now built under the lock and sent outside it, and chunkservers drop tasks when their queues are full instead of back-pressuring the stream. - Heartbeat reconciliation is rewritten to derive state from what a chunkserver reports rather than assuming dispatched tasks succeeded. Over-replication victims are chosen deterministically so concurrent heartbeats cannot each delete their copy. - Three error paths spun at 100% CPU by skipping their sleep; reconnects now use exponential backoff with jitter. - disk_usage measured a hardcoded /home instead of the storage directory, making placement decisions meaningless. - Chunk uploads landed directly on the final filename, so the directory scan could advertise a half-written chunk as a complete replica. They now commit via fsync + rename. - A failed dial was cached as a nil client in the uploader, which would panic on next use. - chunkId is attacker-controlled and was joined straight onto a path; traversal is now rejected. - Fatal startup errors logged and exited 0. - GracefulStop could never complete because heartbeat streams never end; shutdown now falls back to a forced stop. Adds the master flags the other binaries already had, a test suite covering all of the above (the repo had none), CI, and a Makefile. Co-Authored-By: Claude Opus 5 --- .github/workflows/ci.yml | 35 + .gitignore | 5 +- Makefile | 25 + README.md | 85 ++- cmd/chunkserver/main.go | 5 +- cmd/master/main.go | 26 +- internal/chunkserver/server.go | 583 ++++++++++------- internal/chunkserver/utils.go | 41 +- internal/chunkserver/utils_test.go | 64 ++ internal/client/downloader/downloader.go | 218 +++---- internal/client/uploader/chunker.go | 68 +- internal/client/uploader/chunker_test.go | 160 +++++ internal/client/uploader/uploader.go | 162 +++-- internal/integration/dfs_test.go | 366 +++++++++++ internal/master/metastore/metastore.go | 303 +++++++++ internal/master/metastore/metastore_test.go | 183 ++++++ internal/master/server.go | 677 +++++++++++--------- internal/master/server_test.go | 322 ++++++++++ internal/test/testChunker.go | 36 -- 19 files changed, 2568 insertions(+), 796 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 Makefile create mode 100644 internal/chunkserver/utils_test.go create mode 100644 internal/client/uploader/chunker_test.go create mode 100644 internal/integration/dfs_test.go create mode 100644 internal/master/metastore/metastore.go create mode 100644 internal/master/metastore/metastore_test.go create mode 100644 internal/master/server_test.go delete mode 100644 internal/test/testChunker.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..1efc1f0 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,35 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +jobs: + build: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-go@v5 + with: + go-version-file: go.mod + cache: true + + - name: Check formatting + run: | + unformatted=$(gofmt -l .) + if [ -n "$unformatted" ]; then + echo "These files need gofmt:" + echo "$unformatted" + exit 1 + fi + + - name: Build + run: go build ./... + + - name: Vet + run: go vet ./... + + - name: Test + run: go test -race -timeout 10m ./... diff --git a/.gitignore b/.gitignore index fadf293..48c752a 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,3 @@ -libraryTest.go -output.txt bin/ -data/ \ No newline at end of file +data/ +meta/ diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..37c600f --- /dev/null +++ b/Makefile @@ -0,0 +1,25 @@ +.PHONY: build test test-short test-race vet fmt clean + +build: + go build -o bin/dfs-master ./cmd/master + go build -o bin/dfs-chunkserver ./cmd/chunkserver + go build -o bin/dfs-client ./cmd/client + +test: + go test ./... + +# Unit tests only; skips anything that binds a port. +test-short: + go test -short ./... + +test-race: + go test -race -timeout 10m ./... + +vet: + go vet ./... + +fmt: + gofmt -w . + +clean: + rm -rf bin data meta diff --git a/README.md b/README.md index e4f498f..8e46eec 100644 --- a/README.md +++ b/README.md @@ -32,12 +32,14 @@ go build -o bin/dfs-client ./cmd/client | Feature | Description | |---------|-------------| -| File Upload | Chunked upload with automatic allocation | -| File Download | Reassembly from distributed chunks | +| File Upload | Chunked upload written synchronously to every replica | +| File Download | Reassembly from distributed chunks, with replica failover | +| Durable Namespace | Write-ahead log + checkpoints; survives master restarts | | Chunk Replication | Configurable replication factor (default: 2) | | Heartbeat System | Bidirectional streaming for health monitoring | | Dead Server Detection | Automatic detection and replication triggering | | Replica Placement | Storage-aware server selection | +| Orphan Reclamation | Chunks with no namespace entry are collected after a grace period | | Structured Logging | slog-based logging with levels | | Client SDK | High-level API for file operations | | CLI Tool | Command-line interface for put/get operations | @@ -47,7 +49,21 @@ go build -o bin/dfs-client ./cmd/client - Write pipeline (chain replication) - Lease management for primary writes - Checksum verification -- Graceful shutdown handling +- Delete / list / rename, and overwriting an existing file +- TLS and authentication between all components + +### ⚠️ Known Limitations + +- **No checksums.** The `checksum` field exists in the protocol but is always + zero, so bitrot and corrupt transfers go undetected. +- **Single master.** The namespace is durable, but the master is still a single + point of *availability* — there is no standby. Recovery means restarting it + over the same metadata directory. +- **No delete or overwrite.** Re-uploading an existing file name fails with + `AlreadyExists`. +- **No transport security.** All gRPC connections are insecure; anyone who can + reach the cluster can read or register. +- **No RPC deadlines on the client**, so an unresponsive server can hang the CLI. --- @@ -127,16 +143,22 @@ sequenceDiagram participant CS2 as ChunkServer 2 C->>M: AllocateChunk(filename, index) + M->>M: Append to namespace log (fsync) M-->>C: ChunkID + ReplicaServers C->>CS1: UploadChunk(stream) CS1-->>C: Success - - Note over M,CS2: Replication via heartbeat + C->>CS2: UploadChunk(stream) + CS2-->>C: Success + + Note over M,CS2: Heartbeat repairs any replica lost later M->>CS1: ReplicationTask CS1->>CS2: ReplicateChunk CS2-->>CS1: Ack ``` +The client writes every replica before reporting success. Background replication +exists to repair replicas lost *after* the write, not to finish the write itself. + ### Read Sequence ```mermaid @@ -165,7 +187,9 @@ dfs/ │ ├── internal/ # Private implementation │ ├── master/ # Master server logic +│ │ └── metastore/ # Namespace write-ahead log + checkpoints │ ├── chunkserver/ # Chunk server logic +│ ├── integration/ # End-to-end tests over real gRPC │ └── client/ # Client SDK │ ├── dfsclient.go # High-level SDK │ ├── uploader/ # Upload handling @@ -208,19 +232,58 @@ Flags: -dir string Storage directory (default "./data") ``` +### Master Flags + +```bash +./bin/dfs-master [flags] + +Flags: + -port string Address to listen on (default ":8000") + -meta string Namespace log and checkpoint directory (default "./meta") + -replication-factor int Target replicas per chunk (default 2) + -live-threshold duration Heartbeat silence before a server is presumed dead (default 30s) + -orphan-grace duration Delay after startup before reclaiming unknown chunks (default 5m) +``` + --- ## Configuration | Parameter | Default | Description | |-----------|---------|-------------| -| `REPLICATION_FACTOR` | 2 | Number of replicas per chunk | +| `-replication-factor` | 2 | Number of replicas per chunk | | `CHUNK_SIZE` | 64 MB | Size of each chunk | -| `LIVE_THRESHOLD` | 30s | Server considered dead after this | +| `-live-threshold` | 30s | Server considered dead after this | +| `-orphan-grace` | 5m | Startup delay before collecting chunks with no metadata | | Heartbeat Interval | 5s | Chunk server heartbeat frequency | --- +## Metadata Durability + +The master follows the GFS operation-log design, which HDFS mirrors as +`edits` + `fsimage`. Metadata is split into two classes: + +| | Stored where | Recovered how | +|---|---|---| +| **Namespace** (file → ordered chunk list) | Write-ahead log + checkpoints under `-meta` | Replayed at startup | +| **Chunk locations** (chunk → replica servers) | Nowhere — deliberately | Rebuilt from chunkserver registration and heartbeats | + +Every namespace mutation is appended to `namespace.log` and `fsync`'d **before** +the RPC is acknowledged, so an acknowledged write survives a crash. Checkpoints +(`namespace.checkpoint`) are published by atomic rename and exist only to bound +replay time; the log remains the source of truth. A half-written trailing log +record is discarded on recovery — it was never `fsync`'d, so it was never +acknowledged to a client. + +Replica locations are intentionally *not* persisted. Chunkservers are the only +authority on what they actually hold, so storing locations would just produce a +stale copy that disagrees with the disks. This is why a freshly restarted master +waits `-orphan-grace` before deleting chunks it has no metadata for: it must give +every chunkserver time to re-register first. + +--- + ## Data Flow ### Write Path @@ -260,9 +323,15 @@ Chunk servers maintain a bidirectional gRPC stream with the master: # Build all go build ./... -# Run tests +# Run tests (includes end-to-end tests that start real servers) go test ./... +# Unit tests only — skips anything that binds ports +go test -short ./... + +# Race detector +go test -race ./... + # Lint go vet ./... ``` diff --git a/cmd/chunkserver/main.go b/cmd/chunkserver/main.go index 99ed35d..e08622a 100644 --- a/cmd/chunkserver/main.go +++ b/cmd/chunkserver/main.go @@ -1,9 +1,11 @@ package main import ( + "flag" + "os" + "dfs/internal/chunkserver" "dfs/pkg/logger" - "flag" ) func main() { @@ -24,5 +26,6 @@ func main() { if err := cs.Start(); err != nil { logger.Error("Chunk server failed", "error", err) + os.Exit(1) } } diff --git a/cmd/master/main.go b/cmd/master/main.go index d7b5808..68809ef 100644 --- a/cmd/master/main.go +++ b/cmd/master/main.go @@ -1,13 +1,37 @@ package main import ( + "flag" + "os" + "dfs/internal/master" "dfs/pkg/logger" ) func main() { - m := master.NewMasterServer() + cfg := master.DefaultConfig() + + flag.StringVar(&cfg.ListenAddress, "port", cfg.ListenAddress, "address to listen on") + flag.StringVar(&cfg.MetadataDir, "meta", cfg.MetadataDir, "directory for the namespace log and checkpoints") + flag.IntVar(&cfg.ReplicationFactor, "replication-factor", cfg.ReplicationFactor, "target replicas per chunk") + flag.DurationVar(&cfg.LiveThreshold, "live-threshold", cfg.LiveThreshold, "time without a heartbeat before a chunkserver is presumed dead") + flag.DurationVar(&cfg.OrphanGracePeriod, "orphan-grace", cfg.OrphanGracePeriod, "time after startup before reclaiming chunks with no metadata") + flag.Parse() + + if cfg.ReplicationFactor < 1 { + logger.Error("Invalid replication factor", "value", cfg.ReplicationFactor) + os.Exit(2) + } + + m, err := master.NewMasterServer(cfg) + if err != nil { + logger.Error("Master server failed to start", "error", err) + os.Exit(1) + } + defer m.Close() + if err := m.Start(); err != nil { logger.Error("Master server failed", "error", err) + os.Exit(1) } } diff --git a/internal/chunkserver/server.go b/internal/chunkserver/server.go index 5097c41..260f92c 100644 --- a/internal/chunkserver/server.go +++ b/internal/chunkserver/server.go @@ -2,27 +2,47 @@ package chunkserver import ( "context" - "dfs/dfs/chunkpb" - "dfs/dfs/masterpb" - "dfs/pkg/logger" + "errors" "fmt" "io" + "math/rand" "net" "os" "path/filepath" + "strings" "sync" "time" + "dfs/dfs/chunkpb" + "dfs/dfs/masterpb" + "dfs/pkg/logger" + "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" ) +const ( + // tempSuffix marks a chunk that is still being written. Files carrying it are + // invisible to the master, so a partially received chunk is never advertised + // as a usable replica. + tempPrefix = ".partial-" + + heartbeatInterval = 5 * time.Second + refreshInterval = 10 * time.Second + maxBackoff = 30 * time.Second + frameSize = 64 * 1024 + // shutdownGrace bounds how long Stop waits for in-flight RPCs. + shutdownGrace = 3 * time.Second +) + type ChunkServer struct { chunkpb.UnimplementedChunkServiceServer myAddress string masterAddress string storageDir string + grpcServer *grpc.Server + mu sync.Mutex chunks []string replicationTask chan *masterpb.ReplicationTask @@ -35,52 +55,84 @@ func NewChunkServer(myAddress, masterAddress, storageDir string) *ChunkServer { masterAddress: masterAddress, storageDir: storageDir, chunks: make([]string, 0), - replicationTask: make(chan *masterpb.ReplicationTask, 1000), // keep this in mind!! + replicationTask: make(chan *masterpb.ReplicationTask, 1000), deleteTask: make(chan *masterpb.DeleteTask, 1000), } } func (cs *ChunkServer) Start() error { - - if err := os.MkdirAll(cs.storageDir, 0755); err != nil { + if err := os.MkdirAll(cs.storageDir, 0o755); err != nil { return err } + // Clear any temp files left behind by a crash mid-upload. + cs.cleanupPartials() + // Populate the chunk list before announcing ourselves, so the first heartbeat + // reports what we actually hold. + cs.refreshChunksOnce() + lis, err := net.Listen("tcp", cs.myAddress) if err != nil { return err } - grpcServer := grpc.NewServer() - chunkpb.RegisterChunkServiceServer(grpcServer, cs) - // implement a go routine for registering the chunk server // call the master server from there , by creating a connection + cs.grpcServer = grpc.NewServer() + chunkpb.RegisterChunkServiceServer(cs.grpcServer, cs) go cs.runHeartbeat() go cs.replicateAndDeleteTasks() go cs.refreshChunks() - // the server will only start once it is registered on master - // need a thread to monitor delete and replication --> will spawn two threads --> one will do replication , one will delete - logger.Info("ChunkServer started", "address", cs.myAddress) - return grpcServer.Serve(lis) + + logger.Info("ChunkServer started", "address", cs.myAddress, "dir", cs.storageDir) + return cs.grpcServer.Serve(lis) } +// Stop shuts the gRPC server down, letting in-flight transfers finish. A +// transfer that outlasts shutdownGrace is cut off so shutdown cannot hang. +func (cs *ChunkServer) Stop() { + if cs.grpcServer == nil { + return + } + done := make(chan struct{}) + go func() { + cs.grpcServer.GracefulStop() + close(done) + }() + select { + case <-done: + case <-time.After(shutdownGrace): + logger.Warn("Graceful shutdown timed out, closing open transfers", "address", cs.myAddress) + cs.grpcServer.Stop() + <-done + } +} + +// UploadChunk receives a chunk and commits it atomically. +// +// Data lands in a temp file that is fsync'd and then renamed into place. Writing +// directly to the final name would let the background directory scan observe a +// half-written file and advertise it to the master as a complete replica. func (cs *ChunkServer) UploadChunk( stream grpc.ClientStreamingServer[chunkpb.ChunkData, chunkpb.UploadChunkStatus], ) error { + var ( + file *os.File + tmpPath string + chunkID string + finalDst string + ) - //Checksum left - var file *os.File - var chunkId string - - // Ensure file is closed on any exit + // Remove the temp file unless the rename below claimed it. defer func() { if file != nil { - file.Sync() file.Close() } + if tmpPath != "" { + os.Remove(tmpPath) + } }() for { msg, err := stream.Recv() - if err == io.EOF { + if errors.Is(err, io.EOF) { break } if err != nil { @@ -88,31 +140,42 @@ func (cs *ChunkServer) UploadChunk( } if file == nil { - chunkId = msg.GetChunkId() - filePath := filepath.Join(cs.storageDir, chunkId) - - file, err = os.Create(filePath) + chunkID = msg.GetChunkId() + finalDst, err = resolveChunkPath(cs.storageDir, chunkID) if err != nil { return err } - + file, err = os.CreateTemp(cs.storageDir, tempPrefix+"*") + if err != nil { + return fmt.Errorf("create temp chunk: %w", err) + } + tmpPath = file.Name() + } else if msg.GetChunkId() != chunkID { + return fmt.Errorf("stream switched chunk id from %q to %q", chunkID, msg.GetChunkId()) } - _, err = file.Write(msg.GetData()) - // handle partial writes , so maybe just remove the given file for now - if err != nil { - return err + if _, err := file.Write(msg.GetData()); err != nil { + return fmt.Errorf("write chunk %s: %w", chunkID, err) } - } if file == nil { return fmt.Errorf("no data received") } + if err := file.Sync(); err != nil { + return fmt.Errorf("sync chunk %s: %w", chunkID, err) + } + if err := file.Close(); err != nil { + return fmt.Errorf("close chunk %s: %w", chunkID, err) + } + file = nil - cs.mu.Lock() - cs.chunks = append(cs.chunks, chunkId) - cs.mu.Unlock() + if err := os.Rename(tmpPath, finalDst); err != nil { + return fmt.Errorf("commit chunk %s: %w", chunkID, err) + } + tmpPath = "" // now owned by finalDst + + cs.addChunk(chunkID) return stream.SendAndClose(&chunkpb.UploadChunkStatus{Success: true}) } @@ -120,308 +183,360 @@ func (cs *ChunkServer) DownloadChunk( req *chunkpb.DownloadChunkRequest, stream grpc.ServerStreamingServer[chunkpb.ChunkData], ) error { - //Checksum left for now + chunkID := req.GetChunkId() + filePath, err := resolveChunkPath(cs.storageDir, chunkID) + if err != nil { + return err + } - chunkId := req.GetChunkId() - buff := make([]byte, 64*1024) - filePath := filepath.Join(cs.storageDir, chunkId) file, err := os.Open(filePath) - if err != nil { return err } defer file.Close() - var seqNo int32 = 0 + + buff := make([]byte, frameSize) + var seqNo int32 for { n, err := file.Read(buff) - if err == io.EOF { - break - } - if err != nil { - return err + if n > 0 { + if sendErr := stream.Send(&chunkpb.ChunkData{ + SeqNo: seqNo, + ChunkId: chunkID, + Data: buff[:n], + Checksum: 0, + }); sendErr != nil { + return sendErr + } + seqNo++ } - chunk := &chunkpb.ChunkData{ - SeqNo: int32(seqNo), - ChunkId: chunkId, - Data: buff[:n], - Checksum: 0, + if errors.Is(err, io.EOF) { + return nil } - if err := stream.Send(chunk); err != nil { + if err != nil { return err } - seqNo++ } - return nil } -func (cs *ChunkServer) ReplicateChunkToTarget(chunkId, targetAddress string) error { - logger.Debug("Starting replication", "chunk", chunkId, "target", targetAddress) - filePath := filepath.Join(cs.storageDir, chunkId) +// ReplicateChunkToTarget pushes a local chunk to another chunkserver. +func (cs *ChunkServer) ReplicateChunkToTarget(chunkID, targetAddress string) error { + logger.Debug("Starting replication", "chunk", chunkID, "target", targetAddress) + + filePath, err := resolveChunkPath(cs.storageDir, chunkID) + if err != nil { + return err + } file, err := os.Open(filePath) if err != nil { return err } defer file.Close() - buff := make([]byte, 64*1024) conn, err := grpc.NewClient(targetAddress, grpc.WithTransportCredentials(insecure.NewCredentials())) - if err != nil { return err } defer conn.Close() - client := chunkpb.NewChunkServiceClient(conn) - var seqNo int32 = 0 ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() - stream, err := client.UploadChunk(ctx) + + stream, err := chunkpb.NewChunkServiceClient(conn).UploadChunk(ctx) if err != nil { return err } + + buff := make([]byte, frameSize) + var seqNo int32 for { n, err := file.Read(buff) - if err == io.EOF { + if n > 0 { + if sendErr := stream.Send(&chunkpb.ChunkData{ + SeqNo: seqNo, + ChunkId: chunkID, + Data: buff[:n], + Checksum: 0, + }); sendErr != nil { + return sendErr + } + seqNo++ + } + if errors.Is(err, io.EOF) { break } if err != nil { return err } - chunk := &chunkpb.ChunkData{ - SeqNo: int32(seqNo), - ChunkId: chunkId, - Data: buff[:n], - Checksum: 0, - } - if err := stream.Send(chunk); err != nil { - return err - } - seqNo++ } + resp, err := stream.CloseAndRecv() if err != nil { return err } - - if !resp.Success { + if !resp.GetSuccess() { return fmt.Errorf("replication to %s failed", targetAddress) } - logger.Debug("Replication completed", "chunk", chunkId, "target", targetAddress) + logger.Debug("Replication completed", "chunk", chunkID, "target", targetAddress) return nil } -// keeps on trying until registered for now +// registerWithMaster retries until the master accepts the registration. func (cs *ChunkServer) registerWithMaster() { + backoff := time.Second for { - conn, err := grpc.NewClient( - cs.masterAddress, - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - logger.Error("Connection to master failed", "error", err) - time.Sleep(time.Second) + if err := cs.tryRegister(); err != nil { + logger.Error("Registration failed", "error", err, "retry_in", backoff) + time.Sleep(backoff) + backoff = nextBackoff(backoff) continue } + logger.Info("Registered with master") + return + } +} - client := masterpb.NewMasterServiceClient(conn) +// tryRegister is a single registration attempt. It is a separate function so the +// connection and context are released on every attempt rather than accumulating +// deferred cleanups inside a retry loop. +func (cs *ChunkServer) tryRegister() error { + conn, err := grpc.NewClient(cs.masterAddress, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + return err + } + defer conn.Close() - free, err := disk_usage() - if err != nil { - logger.Error("Disk usage check failed", "error", err) - time.Sleep(time.Second) - continue - } - cs.mu.Lock() - chunks := append([]string(nil), cs.chunks...) - cs.mu.Unlock() + free, err := freeStorageMB(cs.storageDir) + if err != nil { + return err + } - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - _, err = client.RegisterChunkServer(ctx, &masterpb.RegisterChunkServerRequest{ - ServerAddress: cs.myAddress, - FreeStorage: free, - Chunks: chunks, - }) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() - conn.Close() + _, err = masterpb.NewMasterServiceClient(conn).RegisterChunkServer(ctx, &masterpb.RegisterChunkServerRequest{ + ServerAddress: cs.myAddress, + FreeStorage: free, + Chunks: cs.snapshotChunks(), + }) + return err +} +func (cs *ChunkServer) runHeartbeat() { + for { + cs.registerWithMaster() + + conn, err := grpc.NewClient(cs.masterAddress, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { - logger.Error("Registration failed", "error", err) + logger.Error("Connection to master failed", "error", err) time.Sleep(time.Second) continue } - logger.Info("Registered with master") - return + client := masterpb.NewMasterServiceClient(conn) + for { + if err := cs.heartbeatSession(client); err != nil { + logger.Error("Heartbeat session ended", "error", err) + break + } + } + conn.Close() } - } -func (cs *ChunkServer) runHeartbeat() { - // create a connection to master and every 5 seconds send the value - var ( - conn *grpc.ClientConn - err error - ) - for { - cs.registerWithMaster() - // now create a connection to master +// heartbeatSession runs one bidirectional heartbeat stream until either +// direction fails, then returns so the caller can re-register and reconnect. +func (cs *ChunkServer) heartbeatSession(client masterpb.MasterServiceClient) error { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + stream, err := client.Heartbeat(ctx) + if err != nil { + return err + } + + var wg sync.WaitGroup + wg.Add(2) + + var recvErr, sendErr error + + go func() { + defer wg.Done() + defer cancel() // unblock the sender when the stream dies for { - conn, err = grpc.NewClient(cs.masterAddress, grpc.WithTransportCredentials(insecure.NewCredentials())) + msg, err := stream.Recv() + if errors.Is(err, io.EOF) { + return + } if err != nil { - fmt.Println("Connection failed to master") - // implement retry mechanism here - if conn != nil { - conn.Close() + recvErr = err + return + } + // Tasks are dropped rather than blocked on when the queues are full. + // Blocking here stops reading the stream, which back-pressures the + // master; the master recomputes outstanding work on the next + // heartbeat, so a dropped task is retried rather than lost. + for _, task := range msg.GetDeleteTasks() { + select { + case cs.deleteTask <- task: + default: + logger.Warn("Delete queue full, dropping task", "chunk", task.GetChunkId()) + } + } + for _, task := range msg.GetReplicationTasks() { + select { + case cs.replicationTask <- task: + default: + logger.Warn("Replication queue full, dropping task", "chunk", task.GetChunkId()) } - continue } - break // connection established } - client := masterpb.NewMasterServiceClient(conn) - for { - ctx := context.Background() + }() - stream, err := client.Heartbeat(ctx) + go func() { + defer wg.Done() + defer cancel() + ticker := time.NewTicker(heartbeatInterval) + defer ticker.Stop() + for { + storage, err := freeStorageMB(cs.storageDir) if err != nil { - logger.Error("Heartbeat RPC failed", "error", err) - break // need to re establish the connection + // Report zero rather than skipping the beat: staying silent would + // have the master declare this server dead. + logger.Error("Disk usage check failed", "error", err) + storage = 0 + } + if err := stream.Send(&masterpb.HeartbeatRequest{ + ServerAddress: cs.myAddress, + FreeStorage: storage, + Chunks: cs.snapshotChunks(), + }); err != nil { + sendErr = err + return + } + select { + case <-ticker.C: + case <-ctx.Done(): + return } - wg := new(sync.WaitGroup) - wg.Add(2) - // now one read thread and one write thread - - // read thread - go func() { - defer wg.Done() - for { - msg, err := stream.Recv() - if err == io.EOF { - break - } - // no use of err here - if err != nil { - logger.Error("Heartbeat read error", "address", cs.myAddress, "error", err) - break // maybe ?? something to think of for now - } - deleteTask := msg.GetDeleteTasks() - replicationTask := msg.GetReplicationTasks() - - // becomes blocking , when the channel is at capacity --> hence the streaming is blocked .. but its fine for now - for _, task := range deleteTask { - cs.deleteTask <- task - } - for _, task := range replicationTask { - cs.replicationTask <- task - } - } - logger.Debug("Exiting heartbeat read thread") - // the implementation seems fine for now - }() - // write thread - go func() { - defer wg.Done() - for { - - storage, err := disk_usage() - if err != nil { - logger.Error("Disk usage check failed") - continue - } - - cs.mu.Lock() - chunks := append([]string(nil), cs.chunks...) - cs.mu.Unlock() - // gather the heartbeat response - heartbeat := &masterpb.HeartbeatRequest{ - ServerAddress: cs.myAddress, - FreeStorage: storage, - Chunks: chunks, - } - - err = stream.Send(heartbeat) - if err != nil { - logger.Error("Heartbeat send failed", "address", cs.myAddress) - // this means the connection is broken , so need to re establish the connection - break - } - time.Sleep(time.Second * 5) // for now , need to check in the master service, what i have configured - // idts there is a need to terminate this loop .. for now!! - } - - logger.Debug("Exiting heartbeat write thread") - }() - - // use wait group , then the thread ends - wg.Wait() - // time.Sleep() } - conn.Close() - } + }() + wg.Wait() + if recvErr != nil { + return recvErr + } + return sendErr } func (cs *ChunkServer) replicateAndDeleteTasks() { - - // replicate task , this is an infinitely running routine , since the channel is never closed go func() { for task := range cs.replicationTask { - // create a connection to a client if err := cs.ReplicateChunkToTarget(task.GetChunkId(), task.GetTargetAddress()); err != nil { logger.Error("Replication failed", "chunk", task.GetChunkId(), "error", err) } } }() - // delete task --> the channel never closes , so this is an infinitely running routine - // althought it is fine , since it is a receiving thread + go func() { for task := range cs.deleteTask { - // delete these files from the disk - filePath := filepath.Join(cs.storageDir, task.GetChunkId()) - if err := os.Remove(filePath); err != nil { - logger.Error("Delete failed", "file", filePath, "error", err) + chunkID := task.GetChunkId() + filePath, err := resolveChunkPath(cs.storageDir, chunkID) + if err != nil { + logger.Error("Delete rejected", "chunk", chunkID, "error", err) continue } - cs.mu.Lock() - index := -1 - - for idx, chunkId := range cs.chunks { - if chunkId == task.GetChunkId() { - index = idx - break - } - } - // use a mutex here maybe ! - // deletion trick ( doesn't preserver order ) - if index != -1 { - cs.chunks[index] = cs.chunks[len(cs.chunks)-1] - cs.chunks = cs.chunks[:len(cs.chunks)-1] + if err := os.Remove(filePath); err != nil && !os.IsNotExist(err) { + logger.Error("Delete failed", "file", filePath, "error", err) + continue } - cs.mu.Unlock() - + cs.removeChunk(chunkID) } }() - } +// refreshChunks keeps the advertised chunk list in step with what is on disk. func (cs *ChunkServer) refreshChunks() { + ticker := time.NewTicker(refreshInterval) + defer ticker.Stop() + for range ticker.C { + cs.refreshChunksOnce() + } +} - for { +func (cs *ChunkServer) refreshChunksOnce() { + entries, err := os.ReadDir(cs.storageDir) + if err != nil { + // Return rather than continue: retrying immediately would spin on a + // persistent error and burn a core. + logger.Error("Chunk refresh failed", "error", err) + return + } - chunks := make([]string, 0) + chunks := make([]string, 0, len(entries)) + for _, entry := range entries { + if entry.IsDir() || strings.HasPrefix(entry.Name(), tempPrefix) { + continue // in-flight uploads are not replicas yet + } + chunks = append(chunks, entry.Name()) + } - filesInDir, err := os.ReadDir(cs.storageDir) - if err != nil { - logger.Error("Chunk refresh failed", "error", err) - continue + cs.mu.Lock() + cs.chunks = chunks + cs.mu.Unlock() +} + +// cleanupPartials removes temp files orphaned by a crash during an upload. +func (cs *ChunkServer) cleanupPartials() { + entries, err := os.ReadDir(cs.storageDir) + if err != nil { + return + } + for _, entry := range entries { + if strings.HasPrefix(entry.Name(), tempPrefix) { + path := filepath.Join(cs.storageDir, entry.Name()) + if err := os.Remove(path); err == nil { + logger.Warn("Removed partial chunk left by a previous run", "file", path) + } + } + } +} + +func (cs *ChunkServer) snapshotChunks() []string { + cs.mu.Lock() + defer cs.mu.Unlock() + return append([]string(nil), cs.chunks...) +} + +func (cs *ChunkServer) addChunk(chunkID string) { + cs.mu.Lock() + defer cs.mu.Unlock() + for _, existing := range cs.chunks { + if existing == chunkID { + return // re-upload of a chunk we already hold } + } + cs.chunks = append(cs.chunks, chunkID) +} - for _, files := range filesInDir { - chunks = append(chunks, files.Name()) +func (cs *ChunkServer) removeChunk(chunkID string) { + cs.mu.Lock() + defer cs.mu.Unlock() + for i, existing := range cs.chunks { + if existing == chunkID { + cs.chunks[i] = cs.chunks[len(cs.chunks)-1] + cs.chunks = cs.chunks[:len(cs.chunks)-1] + return } - cs.mu.Lock() - cs.chunks = chunks - cs.mu.Unlock() - time.Sleep(time.Second * 10) } } + +// nextBackoff doubles d up to maxBackoff and adds jitter so that a fleet of +// chunkservers reconnecting after a master restart does not synchronise. +func nextBackoff(d time.Duration) time.Duration { + next := d * 2 + if next > maxBackoff { + next = maxBackoff + } + return next + time.Duration(rand.Int63n(int64(time.Second))) +} diff --git a/internal/chunkserver/utils.go b/internal/chunkserver/utils.go index 689b832..45eb45c 100644 --- a/internal/chunkserver/utils.go +++ b/internal/chunkserver/utils.go @@ -1,17 +1,42 @@ +//go:build linux || darwin || freebsd || netbsd || openbsd + package chunkserver import ( + "fmt" + "path/filepath" + "strings" "syscall" ) -func disk_usage() (int64, error) { +// freeStorageMB reports the space available in MB on the filesystem backing dir. +// +// This must measure the storage directory rather than a fixed path: chunkservers +// commonly store data outside the home filesystem, and two chunkservers on one +// host that both report the same hardcoded path give the master identical numbers, +// which degrades its storage-aware placement to an arbitrary choice. +func freeStorageMB(dir string) (int64, error) { var stats syscall.Statfs_t - err := syscall.Statfs("/home", &stats) - if err != nil { - return 0, err + if err := syscall.Statfs(dir, &stats); err != nil { + return 0, fmt.Errorf("statfs %s: %w", dir, err) + } + return (int64(stats.Bavail) * int64(stats.Bsize)) / (1024 * 1024), nil +} + +// resolveChunkPath joins chunkID onto storageDir, rejecting IDs that would escape +// it. chunkID arrives over the wire from the master or a peer chunkserver, so a +// value such as "../../etc/passwd" must not be able to address a file outside the +// server's own storage directory. +func resolveChunkPath(storageDir, chunkID string) (string, error) { + if chunkID == "" { + return "", fmt.Errorf("empty chunk id") + } + if strings.ContainsRune(chunkID, filepath.Separator) || strings.Contains(chunkID, "..") { + return "", fmt.Errorf("invalid chunk id %q", chunkID) + } + if strings.HasPrefix(chunkID, ".") { + // Reserved for in-progress writes; see tempSuffix. + return "", fmt.Errorf("invalid chunk id %q", chunkID) } - availableBlocks := stats.Bavail - blockSize := stats.Bsize - storageAvailable := (int64(availableBlocks) * blockSize) / (1024 * 1024) - return int64(storageAvailable), nil + return filepath.Join(storageDir, chunkID), nil } diff --git a/internal/chunkserver/utils_test.go b/internal/chunkserver/utils_test.go new file mode 100644 index 0000000..c13e33d --- /dev/null +++ b/internal/chunkserver/utils_test.go @@ -0,0 +1,64 @@ +package chunkserver + +import ( + "path/filepath" + "strings" + "testing" +) + +// chunkID arrives over the wire, so it must never be able to address a file +// outside the server's storage directory. +func TestResolveChunkPathRejectsTraversal(t *testing.T) { + dir := t.TempDir() + + bad := []string{ + "", + "..", + "../escape", + "../../etc/passwd", + "sub/dir", + "/absolute", + ".partial-sneaky", + "a/../../b", + } + for _, chunkID := range bad { + t.Run(chunkID, func(t *testing.T) { + path, err := resolveChunkPath(dir, chunkID) + if err == nil { + t.Fatalf("resolveChunkPath(%q) = %q, want an error", chunkID, path) + } + }) + } +} + +func TestResolveChunkPathAcceptsNormalIDs(t *testing.T) { + dir := t.TempDir() + + for _, chunkID := range []string{"file.bin_0", "f_12", "a-b_c.txt_3"} { + path, err := resolveChunkPath(dir, chunkID) + if err != nil { + t.Fatalf("resolveChunkPath(%q): %v", chunkID, err) + } + if want := filepath.Join(dir, chunkID); path != want { + t.Fatalf("got %q, want %q", path, want) + } + if !strings.HasPrefix(path, dir) { + t.Fatalf("resolved path %q escaped %q", path, dir) + } + } +} + +func TestFreeStorageMeasuresGivenDirectory(t *testing.T) { + dir := t.TempDir() + free, err := freeStorageMB(dir) + if err != nil { + t.Fatalf("freeStorageMB: %v", err) + } + if free <= 0 { + t.Fatalf("free storage = %d MB, want a positive value", free) + } + + if _, err := freeStorageMB(filepath.Join(dir, "does-not-exist")); err == nil { + t.Fatal("expected an error for a missing directory") + } +} diff --git a/internal/client/downloader/downloader.go b/internal/client/downloader/downloader.go index a2def8d..aaac5c1 100644 --- a/internal/client/downloader/downloader.go +++ b/internal/client/downloader/downloader.go @@ -1,150 +1,154 @@ package downloader import ( - "dfs/internal/client/chunkclient" - "dfs/internal/client/masterclient" - "dfs/pkg/logger" + "errors" "fmt" "io" "os" "path/filepath" - "sort" - "strconv" - "strings" -) -var CHUNK_SIZE = 64 * 1024 * 1024 + "dfs/internal/client/chunkclient" + "dfs/internal/client/masterclient" + "dfs/pkg/logger" +) type Downloader struct { chunkClients map[string]*chunkclient.ChunkClient } -// will contain something , idk rn +func NewDownloader() *Downloader { + return &Downloader{ + chunkClients: make(map[string]*chunkclient.ChunkClient), + } +} + +// client returns a cached connection to addr, dialling if necessary. Failed +// dials are not cached. +func (dc *Downloader) client(addr string) (*chunkclient.ChunkClient, error) { + if existing, ok := dc.chunkClients[addr]; ok { + return existing, nil + } + created, err := chunkclient.NewChunkClient(addr) + if err != nil { + return nil, err + } + dc.chunkClients[addr] = created + return created, nil +} -// check about the meta data from the master , then try download -// return error instead of printing for now +// Download reassembles fileName into downloadPath/fileName. +// +// The file is built in a temporary sibling and renamed into place only after +// every chunk has been written, so an interrupted download never leaves a +// plausible-looking partial file where the real one should be, and re-running a +// download replaces the previous result rather than appending to it. func (dc *Downloader) Download(fileName string, masterClient *masterclient.MasterClient, downloadPath string) error { resp, err := masterClient.GetFileInfo(fileName) if err != nil { return fmt.Errorf("failed to get file metadata for %s: %w", fileName, err) } - // traverse over the file info and download chunks + // The master returns chunks in allocation order. fileInfo := resp.GetFileInfo() - // sort it!! - sort.Slice(fileInfo, func(i, j int) bool { //chunkID = fileName_idx - cid1 := fileInfo[i].ChunkId - cid2 := fileInfo[j].ChunkId - - parts1 := strings.Split(cid1, "_") - parts2 := strings.Split(cid2, "_") + if len(fileInfo) == 0 { + return fmt.Errorf("file %s has no chunks", fileName) + } - // Safety check: ensure we have at least 2 parts - if len(parts1) < 2 || len(parts2) < 2 { - return cid1 < cid2 // Fallback to string comparison - } + if err := os.MkdirAll(downloadPath, 0o755); err != nil { + return fmt.Errorf("failed to create download directory %s: %w", downloadPath, err) + } - // Parse indices from last part - idx1, err1 := strconv.Atoi(parts1[len(parts1)-1]) - idx2, err2 := strconv.Atoi(parts2[len(parts2)-1]) + finalPath := filepath.Join(downloadPath, fileName) + tmp, err := os.CreateTemp(downloadPath, "."+fileName+".part-*") + if err != nil { + return fmt.Errorf("failed to create temporary file in %s: %w", downloadPath, err) + } + tmpPath := tmp.Name() + defer func() { + tmp.Close() + os.Remove(tmpPath) // no-op once the rename below has succeeded + }() - if err1 != nil || err2 != nil { - return cid1 < cid2 // Fallback to string comparison + for _, chunk := range fileInfo { + if err := dc.fetchChunk(chunk.GetChunkId(), chunk.GetReplicaServers(), tmp); err != nil { + return err } + } - return idx1 < idx2 - }) - - if err := os.MkdirAll(downloadPath, 0755); err != nil { - return fmt.Errorf("failed to create download directory %s: %w", downloadPath, err) + if err := tmp.Sync(); err != nil { + return fmt.Errorf("failed to flush %s: %w", tmpPath, err) } - // should be sorted according to smth - for _, chunk := range fileInfo { - // download the chunk - chunkId := chunk.GetChunkId() - replicaServers := chunk.GetReplicaServers() - - // try to download from the replica servers - for _, replicaServer := range replicaServers { - // create a connection - if _, ok := dc.chunkClients[replicaServer]; !ok { - chunkClient, err := chunkclient.NewChunkClient(replicaServer) - if err != nil { - logger.Warn("Connection failed, trying next replica", "replica", replicaServer, "error", err) - continue // this means connection couldn't be formed so need to get data from other server - } - dc.chunkClients[replicaServer] = chunkClient - } + if err := tmp.Close(); err != nil { + return fmt.Errorf("failed to close %s: %w", tmpPath, err) + } + if err := os.Rename(tmpPath, finalPath); err != nil { + return fmt.Errorf("failed to move download into place: %w", err) + } + return nil +} - //download the chunk +// fetchChunk appends one chunk to dst, trying each replica in turn. It returns an +// error when no replica could supply the chunk; silently skipping would produce a +// truncated file and still report success. +func (dc *Downloader) fetchChunk(chunkID string, replicas []string, dst *os.File) error { + if len(replicas) == 0 { + return fmt.Errorf("chunk %s has no replicas", chunkID) + } - chunkClient := dc.chunkClients[replicaServer] - stream, err := chunkClient.DownloadChunk(chunkId) - if err != nil { - logger.Warn("Download start failed", "chunk", chunkId, "replica", replicaServer, "error", err) - continue - } + // Where this chunk starts, so a replica that fails midway can be retried from + // a clean offset rather than leaving a partial write in the file. + start, err := dst.Seek(0, io.SeekCurrent) + if err != nil { + return fmt.Errorf("chunk %s: %w", chunkID, err) + } - // now download from the stream - buff := make([]byte, 0, CHUNK_SIZE) - var success bool = false - for { - chunkData, err := stream.Recv() - if err == io.EOF { - success = true - break - } - if err != nil { - logger.Warn("Stream error", "chunk", chunkId, "error", err) - break - } - buff = append(buff, chunkData.Data...) + var lastErr error + for _, replica := range replicas { + if err := dc.streamChunk(chunkID, replica, dst); err != nil { + logger.Warn("Replica read failed, trying next", "chunk", chunkID, "replica", replica, "error", err) + lastErr = err + if _, seekErr := dst.Seek(start, io.SeekStart); seekErr != nil { + return fmt.Errorf("chunk %s: rewind after failed replica: %w", chunkID, seekErr) } - // err maybe not useful here - if !success { - logger.Warn("Download failed, trying next replica", "chunk", chunkId, "replica", replicaServer) - continue + if truncErr := dst.Truncate(start); truncErr != nil { + return fmt.Errorf("chunk %s: discard partial read: %w", chunkID, truncErr) } - - // open a file in append mode - if err := dc.writeChunkToFile(buff, filepath.Join(downloadPath, fileName)); err != nil { - logger.Warn("Write failed, trying next replica", "chunk", chunkId, "error", err) - continue - } - // check if the file is downloaded - - break - + continue } - - } - return nil -} -func NewDownloader() *Downloader { - return &Downloader{ - chunkClients: make(map[string]*chunkclient.ChunkClient), + return nil } + return fmt.Errorf("chunk %s: all %d replicas failed: %w", chunkID, len(replicas), lastErr) } -func (dc *Downloader) Close() { - for k, chunkClient := range dc.chunkClients { - chunkClient.Close() - delete(dc.chunkClients, k) + +func (dc *Downloader) streamChunk(chunkID, replica string, dst *os.File) error { + conn, err := dc.client(replica) + if err != nil { + return fmt.Errorf("connect: %w", err) } -} -func (dc *Downloader) writeChunkToFile(buff []byte, filePath string) error { - file, err := os.OpenFile(filePath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644) + stream, err := conn.DownloadChunk(chunkID) if err != nil { - return fmt.Errorf("failed to open file: %w", err) + return fmt.Errorf("open stream: %w", err) } - defer file.Close() - if _, err := file.Write(buff); err != nil { - return fmt.Errorf("failed to write: %w", err) + // Frames are written straight through rather than buffered, so memory does not + // scale with chunk size. + for { + chunkData, err := stream.Recv() + if errors.Is(err, io.EOF) { + return nil + } + if err != nil { + return fmt.Errorf("receive: %w", err) + } + if _, err := dst.Write(chunkData.GetData()); err != nil { + return fmt.Errorf("write: %w", err) + } } +} - if err := file.Sync(); err != nil { - return fmt.Errorf("failed to sync: %w", err) +func (dc *Downloader) Close() { + for k, chunkClient := range dc.chunkClients { + chunkClient.Close() + delete(dc.chunkClients, k) } - - return nil } diff --git a/internal/client/uploader/chunker.go b/internal/client/uploader/chunker.go index 115de27..d02b0e2 100644 --- a/internal/client/uploader/chunker.go +++ b/internal/client/uploader/chunker.go @@ -1,74 +1,54 @@ package uploader import ( + "errors" "io" "os" ) -// chunker should --> open the file read 64 mb me break it --> now with that 64 mb , break it into 64 kb blocks and send it -// should i have a struct ?? if i get a file +// Chunker splits a local file into fixed-size chunks for upload. type Chunker struct { file *os.File chunkSize int64 chunkCount int - // add more when needed } func NewChunker(filePath string, chunkSize int64) (*Chunker, error) { - file, err := os.Open(filePath) if err != nil { return nil, err - // handle it } return &Chunker{ - file: file, - chunkSize: chunkSize, - chunkCount: 0, + file: file, + chunkSize: chunkSize, }, nil - } -// add utility functions for reading and stuff - +// NextChunk fills buff with the next chunk and returns it along with the chunk's +// index. It returns io.EOF once the file is exhausted; any other error is a real +// read failure and must not be mistaken for the end of the file. +// +// io.ReadFull is used rather than a bare Read because Read is permitted to return +// fewer bytes than requested. Regular files on Linux happen not to, but pipes, +// FIFOs and network filesystems do, and a short read here would silently produce +// undersized chunks and corrupt the reassembled file. func (chunker *Chunker) NextChunk(buff []byte) ([]byte, int, error) { - - // is it possible ? at a time i am reading - - n, err := chunker.file.Read(buff) - if err != nil { + n, err := io.ReadFull(chunker.file, buff) + switch { + case errors.Is(err, io.EOF): + // Nothing left at all. + return nil, 0, io.EOF + case errors.Is(err, io.ErrUnexpectedEOF): + // Final partial chunk: n bytes are valid. + case err != nil: return nil, 0, err - // handle it do something about it for now - // might also get EOF ?? handle it the same way for now } + chunker.chunkCount++ return buff[:n], chunker.chunkCount - 1, nil } -func (chunker *Chunker) GetChunkAtIndex(index int, buff []byte) ([]byte, error) { - // buff := make([]byte , chunker.chunkSize) - offset := int64(index) * (chunker.chunkSize) - _, err := chunker.file.Seek(offset, io.SeekStart) - if err != nil { - return nil, err - } - n, err := chunker.file.Read(buff) - if err != nil { - return nil, err - } - _, err = chunker.file.Seek(0, io.SeekEnd) - if err != nil { - return nil, err - } - // sets it back to last position!! - return buff[:n], err - -} - -func (chunker *Chunker) Close() { - chunker.file.Close() +// Close releases the underlying file. +func (chunker *Chunker) Close() error { + return chunker.file.Close() } - -// once i change the offset , will it point to the last executed one ?? -// be wary of the offset implementation , read more about the seek function in detail -// test this function out first diff --git a/internal/client/uploader/chunker_test.go b/internal/client/uploader/chunker_test.go new file mode 100644 index 0000000..fc50ed5 --- /dev/null +++ b/internal/client/uploader/chunker_test.go @@ -0,0 +1,160 @@ +package uploader + +import ( + "bytes" + "errors" + "io" + "os" + "path/filepath" + "testing" +) + +func writeTempFile(t *testing.T, data []byte) string { + t.Helper() + path := filepath.Join(t.TempDir(), "input.bin") + if err := os.WriteFile(path, data, 0o644); err != nil { + t.Fatalf("write temp file: %v", err) + } + return path +} + +// readAll drains a chunker and returns the concatenated bytes plus the chunk sizes. +func readAll(t *testing.T, path string, chunkSize int64) ([]byte, []int) { + t.Helper() + chunker, err := NewChunker(path, chunkSize) + if err != nil { + t.Fatalf("NewChunker: %v", err) + } + defer chunker.Close() + + var out []byte + var sizes []int + buf := make([]byte, chunkSize) + for i := 0; ; i++ { + data, idx, err := chunker.NextChunk(buf) + if errors.Is(err, io.EOF) { + break + } + if err != nil { + t.Fatalf("NextChunk: %v", err) + } + if idx != i { + t.Fatalf("chunk index = %d, want %d", idx, i) + } + out = append(out, data...) + sizes = append(sizes, len(data)) + } + return out, sizes +} + +func TestChunkerSplitsAndPreservesBytes(t *testing.T) { + data := bytes.Repeat([]byte("abcdefghij"), 25) // 250 bytes + path := writeTempFile(t, data) + + got, sizes := readAll(t, path, 100) + + if !bytes.Equal(got, data) { + t.Fatal("reassembled bytes differ from the source") + } + want := []int{100, 100, 50} + if len(sizes) != len(want) { + t.Fatalf("chunk sizes = %v, want %v", sizes, want) + } + for i := range want { + if sizes[i] != want[i] { + t.Fatalf("chunk sizes = %v, want %v", sizes, want) + } + } +} + +// A file that is an exact multiple of the chunk size must not produce a trailing +// empty chunk. +func TestChunkerExactMultiple(t *testing.T) { + data := bytes.Repeat([]byte("x"), 200) + path := writeTempFile(t, data) + + got, sizes := readAll(t, path, 100) + + if !bytes.Equal(got, data) { + t.Fatal("reassembled bytes differ from the source") + } + if len(sizes) != 2 { + t.Fatalf("got %d chunks (%v), want 2", len(sizes), sizes) + } +} + +func TestChunkerSmallerThanOneChunk(t *testing.T) { + data := []byte("short") + path := writeTempFile(t, data) + + got, sizes := readAll(t, path, 4096) + + if !bytes.Equal(got, data) { + t.Fatalf("got %q, want %q", got, data) + } + if len(sizes) != 1 || sizes[0] != len(data) { + t.Fatalf("chunk sizes = %v, want [%d]", sizes, len(data)) + } +} + +func TestChunkerEmptyFile(t *testing.T) { + path := writeTempFile(t, nil) + + got, sizes := readAll(t, path, 100) + + if len(got) != 0 || len(sizes) != 0 { + t.Fatalf("empty file produced %d bytes in %v chunks", len(got), sizes) + } +} + +// io.Reader is allowed to return fewer bytes than requested. Regular files on +// Linux generally do not, but pipes and network filesystems do, and a short read +// would otherwise be committed as an undersized chunk and corrupt the file. +func TestChunkerHandlesShortReads(t *testing.T) { + data := bytes.Repeat([]byte("y"), 300) + + pr, pw, err := os.Pipe() + if err != nil { + t.Fatalf("pipe: %v", err) + } + go func() { + defer pw.Close() + // Dribble the data out so each read returns far less than a full chunk. + for i := 0; i < len(data); i += 7 { + end := i + 7 + if end > len(data) { + end = len(data) + } + pw.Write(data[i:end]) + } + }() + + chunker := &Chunker{file: pr, chunkSize: 100} + defer chunker.Close() + + var out []byte + buf := make([]byte, 100) + for { + got, _, err := chunker.NextChunk(buf) + if errors.Is(err, io.EOF) { + break + } + if err != nil { + t.Fatalf("NextChunk: %v", err) + } + if len(got) != 100 && len(out)+len(got) != len(data) { + t.Fatalf("short chunk of %d bytes before end of stream", len(got)) + } + out = append(out, got...) + } + + if !bytes.Equal(out, data) { + t.Fatalf("got %d bytes, want %d", len(out), len(data)) + } +} + +func TestChunkerMissingFile(t *testing.T) { + if _, err := NewChunker(filepath.Join(t.TempDir(), "nope"), 100); err == nil { + t.Fatal("expected an error opening a missing file") + } +} diff --git a/internal/client/uploader/uploader.go b/internal/client/uploader/uploader.go index a588df1..6fdd6d3 100644 --- a/internal/client/uploader/uploader.go +++ b/internal/client/uploader/uploader.go @@ -1,115 +1,141 @@ package uploader import ( + "errors" + "fmt" + "io" + "dfs/dfs/chunkpb" "dfs/internal/client/chunkclient" "dfs/internal/client/masterclient" "dfs/pkg/logger" - "fmt" ) -// provides apis to upload the chunk to chunk server -// should it just accept the string ?? -var CHUNK_SIZE int64 = 64 * 1024 * 1024 -var SEND_CHUNK_SIZE int64 = 64 * 1024 +const ( + // CHUNK_SIZE must match the master's chunk size. + CHUNK_SIZE int64 = 64 * 1024 * 1024 + // SEND_CHUNK_SIZE is the gRPC frame size used to stream a chunk. + SEND_CHUNK_SIZE int64 = 64 * 1024 +) type Uploader struct { chunkServerConn map[string]*chunkclient.ChunkClient } -// get the master conn here as input ?? +func NewUploader() *Uploader { + return &Uploader{ + chunkServerConn: make(map[string]*chunkclient.ChunkClient), + } +} + +// client returns a cached connection to addr, dialling if necessary. A failed +// dial is not cached: storing the nil client would make later lookups believe a +// usable connection exists and panic on first use. +func (uc *Uploader) client(addr string) (*chunkclient.ChunkClient, error) { + if existing, ok := uc.chunkServerConn[addr]; ok { + return existing, nil + } + created, err := chunkclient.NewChunkClient(addr) + if err != nil { + return nil, err + } + uc.chunkServerConn[addr] = created + return created, nil +} + +// Upload streams path to the DFS under fileName. +// +// Every chunk is written to all replicas the master allocated. Writing to only +// one and relying on background re-replication leaves a window where the data +// exists in a single copy while the master's metadata claims otherwise, so a +// single machine failure in that window is silent data loss. func (uc *Uploader) Upload(fileName, path string, masterClient *masterclient.MasterClient) error { - // everytime a upload is called a new master connection is made ?? or should the chunkerOb, err := NewChunker(path, CHUNK_SIZE) - if err != nil { return fmt.Errorf("failed to open file %s: %w", path, err) } defer chunkerOb.Close() + data := make([]byte, CHUNK_SIZE) for { buff, idx, err := chunkerOb.NextChunk(data) - // err can also be eof - if err != nil { - logger.Debug("Finished reading file", "reason", err) + if errors.Is(err, io.EOF) { break } + if err != nil { + // A real read error must fail the upload. Treating it as end-of-file + // would truncate the file and still report success. + return fmt.Errorf("failed to read chunk %d of %s: %w", idx, path, err) + } + resp, err := masterClient.AllocateChunk(fileName, int32(idx)) if err != nil { return fmt.Errorf("failed to allocate chunk %d for file %s: %w", idx, fileName, err) } chunkID := resp.GetChunkId() replicaServers := resp.GetReplicaServers() + if len(replicaServers) == 0 { + return fmt.Errorf("master allocated no replicas for chunk %s", chunkID) + } - // master sends alive servers so assume it works - uploaded := false + written := 0 + var lastErr error for _, replica := range replicaServers { - if _, ok := uc.chunkServerConn[replica]; !ok { - uc.chunkServerConn[replica], err = chunkclient.NewChunkClient(replica) - if err != nil { - logger.Warn("Connection failed, trying next replica", "replica", replica, "error", err) - continue - } - } - - stream, err := uc.chunkServerConn[replica].UploadChunk(chunkID) - if err != nil { - logger.Warn("Upload start failed", "chunk", chunkID, "replica", replica, "error", err) - continue - } - // 64 kb is sent at once - var seqNo int32 = 0 - sendSuccess := true - for offset := 0; offset < len(buff); offset += int(SEND_CHUNK_SIZE) { - end := offset + int(SEND_CHUNK_SIZE) - if end > len(buff) { - end = len(buff) - } - if err := stream.Send(&chunkpb.ChunkData{ - ChunkId: chunkID, - Data: buff[offset:end], - SeqNo: seqNo, - Checksum: 0, // zero for now - }); err != nil { - logger.Warn("Send failed", "chunk", chunkID, "replica", replica, "error", err) - sendSuccess = false - break - } - seqNo++ - } - - if !sendSuccess { - stream.CloseAndRecv() // Cleanup stream, ignore error - continue - } - - resp, err := stream.CloseAndRecv() - if err != nil { - logger.Warn("Upload failed", "chunk", chunkID, "replica", replica, "error", err) - continue - } - - if !resp.GetSuccess() { - logger.Warn("Upload rejected by server", "chunk", chunkID, "replica", replica) + if err := uc.sendChunk(chunkID, replica, buff); err != nil { + logger.Warn("Replica write failed", "chunk", chunkID, "replica", replica, "error", err) + lastErr = err continue } logger.Info("Chunk uploaded", "chunk", chunkID, "replica", replica) - uploaded = true - break - + written++ } - if !uploaded { - return fmt.Errorf("failed to upload chunk %s to any replica", chunkID) + if written < len(replicaServers) { + return fmt.Errorf("chunk %s: wrote %d of %d replicas: %w", + chunkID, written, len(replicaServers), lastErr) } } return nil } -func NewUploader() *Uploader { - return &Uploader{ - chunkServerConn: make(map[string]*chunkclient.ChunkClient), +// sendChunk streams one chunk's bytes to a single replica. +func (uc *Uploader) sendChunk(chunkID, replica string, buff []byte) error { + conn, err := uc.client(replica) + if err != nil { + return fmt.Errorf("connect: %w", err) + } + + stream, err := conn.UploadChunk(chunkID) + if err != nil { + return fmt.Errorf("open stream: %w", err) } + + var seqNo int32 + for offset := 0; offset < len(buff); offset += int(SEND_CHUNK_SIZE) { + end := offset + int(SEND_CHUNK_SIZE) + if end > len(buff) { + end = len(buff) + } + if err := stream.Send(&chunkpb.ChunkData{ + ChunkId: chunkID, + Data: buff[offset:end], + SeqNo: seqNo, + Checksum: 0, // not yet implemented + }); err != nil { + stream.CloseAndRecv() // release the stream; the send error is what matters + return fmt.Errorf("send: %w", err) + } + seqNo++ + } + + resp, err := stream.CloseAndRecv() + if err != nil { + return fmt.Errorf("close stream: %w", err) + } + if !resp.GetSuccess() { + return fmt.Errorf("chunkserver rejected the write") + } + return nil } func (uc *Uploader) Close() { diff --git a/internal/integration/dfs_test.go b/internal/integration/dfs_test.go new file mode 100644 index 0000000..2d69d03 --- /dev/null +++ b/internal/integration/dfs_test.go @@ -0,0 +1,366 @@ +// Package integration exercises a real master and chunkservers over gRPC. +// +// Each test here pins a bug that previously reported success while losing or +// corrupting data, so a regression shows up as a failing assertion rather than a +// quietly wrong file. +package integration + +import ( + "bytes" + "crypto/sha256" + "math/rand" + "net" + "os" + "path/filepath" + "testing" + "time" + + "dfs/internal/chunkserver" + "dfs/internal/client" + "dfs/internal/master" +) + +// freePort reserves an ephemeral port and releases it for the caller to bind. +func freePort(t *testing.T) string { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("reserve port: %v", err) + } + addr := lis.Addr().String() + lis.Close() + return addr +} + +type cluster struct { + masterAddr string + metaDir string + dataDirs []string + csAddrs []string + master *master.MasterServer + servers []*chunkserver.ChunkServer +} + +// startCluster brings up a master and n chunkservers, reusing metaDir and +// dataDirs when they are supplied so a cluster can be restarted in place. +func startCluster(t *testing.T, masterAddr, metaDir string, dataDirs []string) *cluster { + t.Helper() + + cfg := master.DefaultConfig() + cfg.ListenAddress = masterAddr + cfg.MetadataDir = metaDir + cfg.ReplicationFactor = 2 + cfg.LiveThreshold = 10 * time.Second + cfg.OrphanGracePeriod = time.Hour // never collect orphans mid-test + + ms, err := master.NewMasterServer(cfg) + if err != nil { + t.Fatalf("NewMasterServer: %v", err) + } + lis, err := net.Listen("tcp", masterAddr) + if err != nil { + t.Fatalf("listen on %s: %v", masterAddr, err) + } + go ms.Serve(lis) + + c := &cluster{masterAddr: masterAddr, metaDir: metaDir, dataDirs: dataDirs, master: ms} + + for _, dir := range dataDirs { + addr := freePort(t) + cs := chunkserver.NewChunkServer(addr, masterAddr, dir) + go func() { + if err := cs.Start(); err != nil { + t.Logf("chunkserver %s exited: %v", addr, err) + } + }() + c.csAddrs = append(c.csAddrs, addr) + c.servers = append(c.servers, cs) + } + + c.waitForRegistration(t, len(dataDirs)) + return c +} + +// waitForRegistration blocks until the master will accept an allocation, which +// requires every chunkserver to have registered. +func (c *cluster) waitForRegistration(t *testing.T, want int) { + t.Helper() + deadline := time.Now().Add(20 * time.Second) + for time.Now().Before(deadline) { + dfs, err := client.NewDFSClient(c.masterAddr) + if err == nil { + // GetFileInfo on a missing file returns NotFound once the master is + // serving; that is enough to know it is up. + dfs.Close() + } + registered := 0 + for _, addr := range c.csAddrs { + conn, err := net.DialTimeout("tcp", addr, 200*time.Millisecond) + if err == nil { + conn.Close() + registered++ + } + } + if registered == want { + // Chunkservers dial the master on startup; give the registration RPC + // a moment to land. + time.Sleep(1500 * time.Millisecond) + return + } + time.Sleep(200 * time.Millisecond) + } + t.Fatal("chunkservers did not come up in time") +} + +func (c *cluster) stop() { + for _, cs := range c.servers { + cs.Stop() + } + c.master.Stop() + c.master.Close() +} + +// stopMasterOnly simulates a master crash while the chunkservers keep running. +func (c *cluster) stopMasterOnly() { + c.master.Stop() + c.master.Close() +} + +func newCluster(t *testing.T, chunkservers int) *cluster { + t.Helper() + base := t.TempDir() + dirs := make([]string, chunkservers) + for i := range dirs { + dirs[i] = filepath.Join(base, "cs", string(rune('a'+i))) + if err := os.MkdirAll(dirs[i], 0o755); err != nil { + t.Fatalf("mkdir: %v", err) + } + } + metaDir := filepath.Join(base, "meta") + c := startCluster(t, freePort(t), metaDir, dirs) + t.Cleanup(c.stop) + return c +} + +func randomFile(t *testing.T, size int) (string, []byte) { + t.Helper() + data := make([]byte, size) + rng := rand.New(rand.NewSource(1)) + rng.Read(data) + path := filepath.Join(t.TempDir(), "source.bin") + if err := os.WriteFile(path, data, 0o644); err != nil { + t.Fatalf("write source: %v", err) + } + return path, data +} + +func TestUploadDownloadRoundTrip(t *testing.T) { + if testing.Short() { + t.Skip("starts real servers") + } + c := newCluster(t, 2) + + srcPath, want := randomFile(t, 3<<20) // 3 MB + dfs, err := client.NewDFSClient(c.masterAddr) + if err != nil { + t.Fatalf("NewDFSClient: %v", err) + } + defer dfs.Close() + + if err := dfs.Put("round.bin", srcPath); err != nil { + t.Fatalf("Put: %v", err) + } + + outDir := t.TempDir() + if err := dfs.Get("round.bin", outDir); err != nil { + t.Fatalf("Get: %v", err) + } + + got, err := os.ReadFile(filepath.Join(outDir, "round.bin")) + if err != nil { + t.Fatalf("read downloaded file: %v", err) + } + if sha256.Sum256(got) != sha256.Sum256(want) { + t.Fatalf("downloaded %d bytes, want %d, and contents differ", len(got), len(want)) + } +} + +// The uploader used to stop after the first successful replica, leaving the data +// in one copy while the master's metadata claimed the full replication factor. +func TestUploadWritesEveryReplicaSynchronously(t *testing.T) { + if testing.Short() { + t.Skip("starts real servers") + } + c := newCluster(t, 2) + + srcPath, _ := randomFile(t, 1<<20) + dfs, err := client.NewDFSClient(c.masterAddr) + if err != nil { + t.Fatalf("NewDFSClient: %v", err) + } + defer dfs.Close() + + if err := dfs.Put("replicated.bin", srcPath); err != nil { + t.Fatalf("Put: %v", err) + } + + // Checked immediately: background re-replication must not be what makes this + // pass, otherwise there is a window where the data exists only once. + for _, dir := range c.dataDirs { + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("read %s: %v", dir, err) + } + found := false + for _, e := range entries { + if e.Name() == "replicated.bin_0" { + found = true + } + } + if !found { + t.Fatalf("chunk missing from %s right after upload; only some replicas were written", dir) + } + } +} + +// Downloading twice into the same directory used to append to the previous +// result, silently doubling the file. +func TestRepeatedDownloadDoesNotAppend(t *testing.T) { + if testing.Short() { + t.Skip("starts real servers") + } + c := newCluster(t, 2) + + srcPath, want := randomFile(t, 1<<20) + dfs, err := client.NewDFSClient(c.masterAddr) + if err != nil { + t.Fatalf("NewDFSClient: %v", err) + } + defer dfs.Close() + + if err := dfs.Put("twice.bin", srcPath); err != nil { + t.Fatalf("Put: %v", err) + } + + outDir := t.TempDir() + for i := 0; i < 3; i++ { + if err := dfs.Get("twice.bin", outDir); err != nil { + t.Fatalf("Get #%d: %v", i+1, err) + } + got, err := os.ReadFile(filepath.Join(outDir, "twice.bin")) + if err != nil { + t.Fatalf("read after Get #%d: %v", i+1, err) + } + if len(got) != len(want) { + t.Fatalf("after %d downloads the file is %d bytes, want %d", i+1, len(got), len(want)) + } + if !bytes.Equal(got, want) { + t.Fatalf("contents differ after %d downloads", i+1) + } + } +} + +// A download that cannot reach any replica used to return nil and exit zero, +// leaving no file behind. +func TestDownloadFailsWhenNoReplicaIsReachable(t *testing.T) { + if testing.Short() { + t.Skip("starts real servers") + } + c := newCluster(t, 2) + + srcPath, _ := randomFile(t, 512<<10) + dfs, err := client.NewDFSClient(c.masterAddr) + if err != nil { + t.Fatalf("NewDFSClient: %v", err) + } + defer dfs.Close() + + if err := dfs.Put("gone.bin", srcPath); err != nil { + t.Fatalf("Put: %v", err) + } + + for _, cs := range c.servers { + cs.Stop() + } + + outDir := t.TempDir() + err = dfs.Get("gone.bin", outDir) + if err == nil { + t.Fatal("Get reported success with every replica unreachable") + } + if _, statErr := os.Stat(filepath.Join(outDir, "gone.bin")); statErr == nil { + t.Fatal("a partial file was left behind by a failed download") + } +} + +// The master used to keep its namespace only in memory, so a restart left every +// chunk stranded on disk and every file permanently unreachable. +func TestNamespaceSurvivesMasterRestart(t *testing.T) { + if testing.Short() { + t.Skip("starts real servers") + } + c := newCluster(t, 2) + + srcPath, want := randomFile(t, 2<<20) + dfs, err := client.NewDFSClient(c.masterAddr) + if err != nil { + t.Fatalf("NewDFSClient: %v", err) + } + if err := dfs.Put("durable.bin", srcPath); err != nil { + t.Fatalf("Put: %v", err) + } + dfs.Close() + + // Crash the master, leaving the chunkservers running. + c.stopMasterOnly() + time.Sleep(500 * time.Millisecond) + + // Bring a new master up over the same metadata directory. + cfg := master.DefaultConfig() + cfg.ListenAddress = c.masterAddr + cfg.MetadataDir = c.metaDir + cfg.ReplicationFactor = 2 + cfg.LiveThreshold = 10 * time.Second + cfg.OrphanGracePeriod = time.Hour + + restarted, err := master.NewMasterServer(cfg) + if err != nil { + t.Fatalf("restart master: %v", err) + } + lis, err := net.Listen("tcp", c.masterAddr) + if err != nil { + t.Fatalf("rebind master: %v", err) + } + go restarted.Serve(lis) + t.Cleanup(func() { restarted.Stop(); restarted.Close() }) + c.master = restarted + + // Chunkservers re-register on their own; wait for locations to be relearned. + var got []byte + deadline := time.Now().Add(30 * time.Second) + for time.Now().Before(deadline) { + dfs2, err := client.NewDFSClient(c.masterAddr) + if err != nil { + time.Sleep(500 * time.Millisecond) + continue + } + outDir := t.TempDir() + err = dfs2.Get("durable.bin", outDir) + dfs2.Close() + if err == nil { + got, err = os.ReadFile(filepath.Join(outDir, "durable.bin")) + if err != nil { + t.Fatalf("read recovered file: %v", err) + } + break + } + time.Sleep(500 * time.Millisecond) + } + + if got == nil { + t.Fatal("file was still unreachable after the master restarted") + } + if sha256.Sum256(got) != sha256.Sum256(want) { + t.Fatal("file recovered after restart does not match the original") + } +} diff --git a/internal/master/metastore/metastore.go b/internal/master/metastore/metastore.go new file mode 100644 index 0000000..fed2857 --- /dev/null +++ b/internal/master/metastore/metastore.go @@ -0,0 +1,303 @@ +// Package metastore provides crash-durable storage for the master's namespace. +// +// The design follows GFS's operation log / checkpoint split, which HDFS mirrors +// as edits / fsimage: +// +// - The namespace (file -> ordered chunk list) is PERSISTENT. Every mutation is +// appended to a write-ahead log and fsync'd before the originating RPC is +// acknowledged, so an acknowledged write survives a crash. +// +// - Chunk replica locations are SOFT STATE and are deliberately NOT stored here. +// Chunkservers are the only authority on what they actually hold, so the master +// rebuilds locations from registration and heartbeats on every boot. Persisting +// them would only create a stale copy that disagrees with reality. +// +// Recovery loads the newest checkpoint and replays the log records written after +// it. Checkpoints exist purely to bound replay time; the log is the source of truth. +package metastore + +import ( + "bufio" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "sync" +) + +const ( + walName = "namespace.log" + checkpointName = "namespace.checkpoint" + checkpointTempName = "namespace.checkpoint.tmp" +) + +// RecordType identifies a namespace mutation in the log. +type RecordType string + +const ( + // RecordAllocateChunk appends ChunkID to FileName's chunk list, creating the + // file if it does not exist. + RecordAllocateChunk RecordType = "allocate_chunk" +) + +// Record is a single durable namespace mutation. +type Record struct { + Seq uint64 `json:"seq"` + Type RecordType `json:"type"` + FileName string `json:"file,omitempty"` + ChunkID string `json:"chunk,omitempty"` +} + +// Snapshot is the reconstructed namespace: file name -> ordered chunk IDs. +// It carries no replica locations by design; see the package comment. +type Snapshot struct { + Seq uint64 `json:"seq"` + Files map[string][]string `json:"files"` +} + +// Store appends namespace mutations to a write-ahead log and periodically folds +// that log into a checkpoint. It is safe for concurrent use. +type Store struct { + dir string + + mu sync.Mutex + wal *os.File + seq uint64 + sinceCheckpnt int +} + +// CheckpointEvery is the number of logged records after which Store reports that +// a checkpoint is due. Checkpointing is driven by the caller because only the +// caller can produce a consistent view of the namespace. +const CheckpointEvery = 1000 + +// Open recovers the namespace from dir and returns a Store ready for appends. +// dir is created if it does not exist. The returned Snapshot is never nil. +func Open(dir string) (*Store, *Snapshot, error) { + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, nil, fmt.Errorf("create metadata dir %s: %w", dir, err) + } + + snap, err := loadCheckpoint(filepath.Join(dir, checkpointName)) + if err != nil { + return nil, nil, err + } + + replayed, err := replayWAL(filepath.Join(dir, walName), snap) + if err != nil { + return nil, nil, err + } + + wal, err := os.OpenFile(filepath.Join(dir, walName), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644) + if err != nil { + return nil, nil, fmt.Errorf("open write-ahead log: %w", err) + } + + return &Store{ + dir: dir, + wal: wal, + seq: snap.Seq, + sinceCheckpnt: replayed, + }, snap, nil +} + +func loadCheckpoint(path string) (*Snapshot, error) { + empty := &Snapshot{Files: make(map[string][]string)} + + data, err := os.ReadFile(path) + if os.IsNotExist(err) { + return empty, nil + } + if err != nil { + return nil, fmt.Errorf("read checkpoint: %w", err) + } + + var snap Snapshot + if err := json.Unmarshal(data, &snap); err != nil { + // A checkpoint is only ever published by atomic rename, so a corrupt one + // means the disk lied to us. Refuse to start rather than silently serving + // a truncated namespace. + return nil, fmt.Errorf("corrupt checkpoint %s: %w", path, err) + } + if snap.Files == nil { + snap.Files = make(map[string][]string) + } + return &snap, nil +} + +// replayWAL applies log records newer than the snapshot, returning how many were +// applied. A torn trailing record is expected after a crash and is discarded: it +// was never fsync'd, so it was never acknowledged to a client. +func replayWAL(path string, snap *Snapshot) (int, error) { + file, err := os.Open(path) + if os.IsNotExist(err) { + return 0, nil + } + if err != nil { + return 0, fmt.Errorf("open write-ahead log: %w", err) + } + defer file.Close() + + applied := 0 + scanner := bufio.NewScanner(file) + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + + for scanner.Scan() { + line := scanner.Bytes() + if len(line) == 0 { + continue + } + var rec Record + if err := json.Unmarshal(line, &rec); err != nil { + // Torn tail: stop here and keep everything before it. + break + } + if rec.Seq <= snap.Seq { + continue // already folded into the checkpoint + } + apply(snap, rec) + applied++ + } + if err := scanner.Err(); err != nil && err != io.EOF { + return applied, fmt.Errorf("read write-ahead log: %w", err) + } + return applied, nil +} + +func apply(snap *Snapshot, rec Record) { + switch rec.Type { + case RecordAllocateChunk: + snap.Files[rec.FileName] = append(snap.Files[rec.FileName], rec.ChunkID) + } + if rec.Seq > snap.Seq { + snap.Seq = rec.Seq + } +} + +// AppendAllocateChunk durably records that chunkID belongs to fileName. It +// returns only after the record is on stable storage, so callers may treat a nil +// return as a commit. +func (s *Store) AppendAllocateChunk(fileName, chunkID string) error { + return s.append(Record{ + Type: RecordAllocateChunk, + FileName: fileName, + ChunkID: chunkID, + }) +} + +func (s *Store) append(rec Record) error { + s.mu.Lock() + defer s.mu.Unlock() + + s.seq++ + rec.Seq = s.seq + + line, err := json.Marshal(rec) + if err != nil { + s.seq-- + return fmt.Errorf("encode log record: %w", err) + } + line = append(line, '\n') + + if _, err := s.wal.Write(line); err != nil { + return fmt.Errorf("write log record: %w", err) + } + if err := s.wal.Sync(); err != nil { + return fmt.Errorf("sync log record: %w", err) + } + + s.sinceCheckpnt++ + return nil +} + +// CheckpointDue reports whether enough records have accumulated to justify +// folding the log into a new checkpoint. +func (s *Store) CheckpointDue() bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.sinceCheckpnt >= CheckpointEvery +} + +// Checkpoint atomically publishes files as the new checkpoint and truncates the +// log. files must be a consistent view of the namespace captured by the caller +// while holding its own lock. +// +// The sequence is: write a temp file, fsync it, rename it into place, fsync the +// directory, then truncate the log. A crash at any point is safe because records +// at or below the checkpoint's sequence number are skipped during replay. +func (s *Store) Checkpoint(files map[string][]string) error { + s.mu.Lock() + defer s.mu.Unlock() + + snap := Snapshot{Seq: s.seq, Files: files} + data, err := json.Marshal(&snap) + if err != nil { + return fmt.Errorf("encode checkpoint: %w", err) + } + + tmpPath := filepath.Join(s.dir, checkpointTempName) + tmp, err := os.OpenFile(tmpPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644) + if err != nil { + return fmt.Errorf("create checkpoint: %w", err) + } + if _, err := tmp.Write(data); err != nil { + tmp.Close() + os.Remove(tmpPath) + return fmt.Errorf("write checkpoint: %w", err) + } + if err := tmp.Sync(); err != nil { + tmp.Close() + os.Remove(tmpPath) + return fmt.Errorf("sync checkpoint: %w", err) + } + if err := tmp.Close(); err != nil { + os.Remove(tmpPath) + return fmt.Errorf("close checkpoint: %w", err) + } + + if err := os.Rename(tmpPath, filepath.Join(s.dir, checkpointName)); err != nil { + os.Remove(tmpPath) + return fmt.Errorf("publish checkpoint: %w", err) + } + if err := syncDir(s.dir); err != nil { + return fmt.Errorf("sync metadata dir: %w", err) + } + + // Records up to snap.Seq are now durable in the checkpoint, so the log can + // start over. Anything appended after this point gets a higher sequence. + if err := s.wal.Truncate(0); err != nil { + return fmt.Errorf("truncate log: %w", err) + } + if _, err := s.wal.Seek(0, io.SeekStart); err != nil { + return fmt.Errorf("rewind log: %w", err) + } + if err := s.wal.Sync(); err != nil { + return fmt.Errorf("sync truncated log: %w", err) + } + + s.sinceCheckpnt = 0 + return nil +} + +// syncDir fsyncs a directory so a rename within it is durable. +func syncDir(dir string) error { + d, err := os.Open(dir) + if err != nil { + return err + } + defer d.Close() + return d.Sync() +} + +// Close flushes and releases the log. +func (s *Store) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + if s.wal == nil { + return nil + } + err := s.wal.Close() + s.wal = nil + return err +} diff --git a/internal/master/metastore/metastore_test.go b/internal/master/metastore/metastore_test.go new file mode 100644 index 0000000..f17aa29 --- /dev/null +++ b/internal/master/metastore/metastore_test.go @@ -0,0 +1,183 @@ +package metastore + +import ( + "os" + "path/filepath" + "reflect" + "testing" +) + +func TestOpenEmptyDir(t *testing.T) { + store, snap, err := Open(t.TempDir()) + if err != nil { + t.Fatalf("Open: %v", err) + } + defer store.Close() + + if len(snap.Files) != 0 { + t.Fatalf("expected empty namespace, got %v", snap.Files) + } +} + +func TestAppendSurvivesReopen(t *testing.T) { + dir := t.TempDir() + + store, _, err := Open(dir) + if err != nil { + t.Fatalf("Open: %v", err) + } + for _, chunk := range []string{"a.bin_0", "a.bin_1"} { + if err := store.AppendAllocateChunk("a.bin", chunk); err != nil { + t.Fatalf("AppendAllocateChunk: %v", err) + } + } + if err := store.AppendAllocateChunk("b.bin", "b.bin_0"); err != nil { + t.Fatalf("AppendAllocateChunk: %v", err) + } + store.Close() + + // Reopen as a restarted master would. + reopened, snap, err := Open(dir) + if err != nil { + t.Fatalf("reopen: %v", err) + } + defer reopened.Close() + + want := map[string][]string{ + "a.bin": {"a.bin_0", "a.bin_1"}, + "b.bin": {"b.bin_0"}, + } + if !reflect.DeepEqual(snap.Files, want) { + t.Fatalf("namespace after restart = %v, want %v", snap.Files, want) + } +} + +func TestChunkOrderIsPreserved(t *testing.T) { + dir := t.TempDir() + store, _, err := Open(dir) + if err != nil { + t.Fatalf("Open: %v", err) + } + // Deliberately not lexicographic: chunk 10 sorts before chunk 2 as a string, + // so a namespace that recovered in the wrong order would reassemble corrupt + // files. + for _, chunk := range []string{"f_0", "f_1", "f_2", "f_10", "f_11"} { + if err := store.AppendAllocateChunk("f", chunk); err != nil { + t.Fatalf("append: %v", err) + } + } + store.Close() + + _, snap, err := Open(dir) + if err != nil { + t.Fatalf("reopen: %v", err) + } + want := []string{"f_0", "f_1", "f_2", "f_10", "f_11"} + if !reflect.DeepEqual(snap.Files["f"], want) { + t.Fatalf("chunk order = %v, want %v", snap.Files["f"], want) + } +} + +func TestCheckpointTruncatesLogAndPreservesState(t *testing.T) { + dir := t.TempDir() + store, _, err := Open(dir) + if err != nil { + t.Fatalf("Open: %v", err) + } + for _, chunk := range []string{"f_0", "f_1"} { + if err := store.AppendAllocateChunk("f", chunk); err != nil { + t.Fatalf("append: %v", err) + } + } + + if err := store.Checkpoint(map[string][]string{"f": {"f_0", "f_1"}}); err != nil { + t.Fatalf("Checkpoint: %v", err) + } + + info, err := os.Stat(filepath.Join(dir, walName)) + if err != nil { + t.Fatalf("stat log: %v", err) + } + if info.Size() != 0 { + t.Fatalf("log should be truncated after checkpoint, size = %d", info.Size()) + } + + // Records written after the checkpoint must still be replayed on top of it. + if err := store.AppendAllocateChunk("f", "f_2"); err != nil { + t.Fatalf("append after checkpoint: %v", err) + } + store.Close() + + _, snap, err := Open(dir) + if err != nil { + t.Fatalf("reopen: %v", err) + } + want := []string{"f_0", "f_1", "f_2"} + if !reflect.DeepEqual(snap.Files["f"], want) { + t.Fatalf("after checkpoint + replay = %v, want %v", snap.Files["f"], want) + } +} + +// A crash can leave a half-written trailing record. It was never fsync'd, so it +// was never acknowledged, and dropping it must not disturb the records before it. +func TestTornTrailingRecordIsDiscarded(t *testing.T) { + dir := t.TempDir() + store, _, err := Open(dir) + if err != nil { + t.Fatalf("Open: %v", err) + } + if err := store.AppendAllocateChunk("f", "f_0"); err != nil { + t.Fatalf("append: %v", err) + } + store.Close() + + walPath := filepath.Join(dir, walName) + f, err := os.OpenFile(walPath, os.O_WRONLY|os.O_APPEND, 0o644) + if err != nil { + t.Fatalf("open log: %v", err) + } + if _, err := f.WriteString(`{"seq":2,"type":"allocate_chunk","fi`); err != nil { + t.Fatalf("write torn record: %v", err) + } + f.Close() + + _, snap, err := Open(dir) + if err != nil { + t.Fatalf("reopen with torn tail: %v", err) + } + if want := []string{"f_0"}; !reflect.DeepEqual(snap.Files["f"], want) { + t.Fatalf("after torn tail = %v, want %v", snap.Files["f"], want) + } +} + +func TestCorruptCheckpointIsRefused(t *testing.T) { + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, checkpointName), []byte("{not json"), 0o644); err != nil { + t.Fatalf("write checkpoint: %v", err) + } + // Serving a truncated namespace would look like data loss to clients, so + // starting must fail loudly instead. + if _, _, err := Open(dir); err == nil { + t.Fatal("expected Open to fail on a corrupt checkpoint") + } +} + +func TestCheckpointDue(t *testing.T) { + store, _, err := Open(t.TempDir()) + if err != nil { + t.Fatalf("Open: %v", err) + } + defer store.Close() + + if store.CheckpointDue() { + t.Fatal("no checkpoint should be due on a fresh store") + } + for i := 0; i < CheckpointEvery; i++ { + if err := store.AppendAllocateChunk("f", "c"); err != nil { + t.Fatalf("append: %v", err) + } + } + if !store.CheckpointDue() { + t.Fatalf("checkpoint should be due after %d records", CheckpointEvery) + } +} diff --git a/internal/master/server.go b/internal/master/server.go index 187ae4f..68d0ef4 100644 --- a/internal/master/server.go +++ b/internal/master/server.go @@ -1,12 +1,7 @@ package master -// TODO : if the master is down , the chunk servers should shift to registerTomaster mode and keep retrying - import ( "context" - "dfs/dfs/masterpb" - "dfs/pkg/logger" - "fmt" "io" "net" "sort" @@ -14,22 +9,67 @@ import ( "sync" "time" + "dfs/dfs/masterpb" + "dfs/internal/master/metastore" + "dfs/pkg/logger" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) -// hardcoding replication factor for now -var REPLICATION_FACTOR int = 2 +const ( + // CHUNK_SIZE is the maximum bytes in one chunk. + CHUNK_SIZE int64 = 64 * 1024 * 1024 + // CHUNK_SIZE_MB is CHUNK_SIZE expressed in MB, matching the units chunkservers + // report free storage in. + CHUNK_SIZE_MB int64 = 64 +) + +// Config holds the master's tunables. +type Config struct { + // ListenAddress is the host:port the gRPC server binds to. + ListenAddress string + // MetadataDir holds the namespace log and checkpoints. + MetadataDir string + // ReplicationFactor is the target number of replicas per chunk. + ReplicationFactor int + // LiveThreshold is how long a chunkserver may go without a heartbeat before + // it is presumed dead and its chunks are re-replicated. + LiveThreshold time.Duration + // OrphanGracePeriod is how long after boot the master waits before deleting + // chunks it has no metadata for. This must comfortably exceed the time for + // every chunkserver to re-register, otherwise a restart would delete live data. + OrphanGracePeriod time.Duration +} -const CHUNK_SIZE int64 = 64 * 1024 * 1024 // 64MB in bytes -const CHUNK_SIZE_MB int64 = 64 // 64MB (for storage comparison) -const LIVE_THRESHOLD = 30 * time.Second // Server considered dead after this +// DefaultConfig returns the configuration used when no flags are supplied. +func DefaultConfig() Config { + return Config{ + ListenAddress: ":8000", + MetadataDir: "./meta", + ReplicationFactor: 2, + LiveThreshold: 30 * time.Second, + OrphanGracePeriod: 5 * time.Minute, + } +} type MasterServer struct { masterpb.UnimplementedMasterServiceServer - mu sync.Mutex - ChunkServers map[string]*ChunkServerInfo - Files map[string]*FileMetaData - Chunks map[string]*ChunkInfo + + cfg Config + store *metastore.Store + grpcServer *grpc.Server + + // bootTime gates orphan collection so a freshly restarted master does not + // delete chunks belonging to servers that have not re-registered yet. + bootTime time.Time + + mu sync.Mutex + ChunkServers map[string]*ChunkServerInfo + Files map[string]*FileMetaData + Chunks map[string]*ChunkInfo + // ReplicationWorks maps a source chunkserver to the copies it should push. ReplicationWorks map[string][]*ReplicationWork } @@ -48,6 +88,7 @@ type FileMetaData struct { type ChunkInfo struct { ChunkID string FileName string + // Replicas is soft state rebuilt from heartbeats; it is never persisted. Replicas map[string]bool } @@ -56,406 +97,459 @@ type ReplicationWork struct { ChunkID string } -func NewMasterServer() *MasterServer { - return &MasterServer{ +// NewMasterServer recovers the namespace from cfg.MetadataDir and returns a +// master ready to serve. Replica locations start empty and are relearned from +// chunkserver registrations and heartbeats. +func NewMasterServer(cfg Config) (*MasterServer, error) { + store, snap, err := metastore.Open(cfg.MetadataDir) + if err != nil { + return nil, err + } + + ms := &MasterServer{ + cfg: cfg, + store: store, + bootTime: time.Now().UTC(), ChunkServers: make(map[string]*ChunkServerInfo), Files: make(map[string]*FileMetaData), Chunks: make(map[string]*ChunkInfo), ReplicationWorks: make(map[string][]*ReplicationWork), } + + for fileName, chunkIDs := range snap.Files { + meta := &FileMetaData{FileName: fileName, Chunks: append([]string(nil), chunkIDs...)} + ms.Files[fileName] = meta + for _, chunkID := range chunkIDs { + ms.Chunks[chunkID] = &ChunkInfo{ + ChunkID: chunkID, + FileName: fileName, + Replicas: make(map[string]bool), + } + } + } + + logger.Info("Namespace recovered", "files", len(ms.Files), "chunks", len(ms.Chunks)) + return ms, nil } func (ms *MasterServer) Start() error { - lis, err := net.Listen("tcp", ":8000") + lis, err := net.Listen("tcp", ms.cfg.ListenAddress) if err != nil { return err } - grpcServer := grpc.NewServer() - masterpb.RegisterMasterServiceServer(grpcServer, ms) - logger.Info("MasterServer started", "port", 8000) + return ms.Serve(lis) +} + +// Serve runs the gRPC server on lis until Stop is called. +func (ms *MasterServer) Serve(lis net.Listener) error { + ms.grpcServer = grpc.NewServer() + masterpb.RegisterMasterServiceServer(ms.grpcServer, ms) + logger.Info("MasterServer started", + "address", lis.Addr().String(), + "replication_factor", ms.cfg.ReplicationFactor, + "metadata_dir", ms.cfg.MetadataDir) + go ms.detectDeadChunkServer() - return grpcServer.Serve(lis) + go ms.checkpointLoop() + return ms.grpcServer.Serve(lis) +} + +// shutdownGrace is how long Stop waits for in-flight RPCs before forcing the +// server down. +const shutdownGrace = 3 * time.Second + +// Stop shuts the gRPC server down. +// +// Heartbeats are long-lived bidirectional streams that only end when a +// chunkserver goes away, so GracefulStop on its own would block indefinitely. +// Unary RPCs are given shutdownGrace to finish, then the remaining streams are +// forced closed; chunkservers reconnect on their own. +func (ms *MasterServer) Stop() { + if ms.grpcServer == nil { + return + } + done := make(chan struct{}) + go func() { + ms.grpcServer.GracefulStop() + close(done) + }() + select { + case <-done: + case <-time.After(shutdownGrace): + logger.Warn("Graceful shutdown timed out, closing open streams") + ms.grpcServer.Stop() + <-done + } +} +// Close releases the namespace log. Call Stop first. +func (ms *MasterServer) Close() error { + return ms.store.Close() } func (ms *MasterServer) RegisterChunkServer(ctx context.Context, req *masterpb.RegisterChunkServerRequest) (*masterpb.RegisterChunkServerResponse, error) { + address := req.GetServerAddress() + if address == "" { + return nil, status.Error(codes.InvalidArgument, "server_address is required") + } - chunkMap := make(map[string]bool) + reported := make(map[string]bool, len(req.GetChunks())) for _, chunk := range req.GetChunks() { - chunkMap[chunk] = true + reported[chunk] = true } - serverInfo := &ChunkServerInfo{ - Address: req.GetServerAddress(), + ms.mu.Lock() + defer ms.mu.Unlock() + + ms.ChunkServers[address] = &ChunkServerInfo{ + Address: address, FreeStorage: req.GetFreeStorage(), - Chunks: chunkMap, + Chunks: reported, LastHeartbeat: time.Now().UTC(), } - ms.mu.Lock() - defer ms.mu.Unlock() - ms.ChunkServers[serverInfo.Address] = serverInfo - for _, chunk := range req.GetChunks() { - if value, ok := ms.Chunks[chunk]; ok { - value.Replicas[req.GetServerAddress()] = true - } else { - chunkInfo := &ChunkInfo{ - ChunkID: chunk, - Replicas: make(map[string]bool), - } - chunkInfo.Replicas[req.GetServerAddress()] = true - ms.Chunks[chunk] = chunkInfo + // Only record locations for chunks the namespace already knows about. Chunks + // the master has no metadata for are orphans (for example, left behind by a + // deleted file) and must not be resurrected into the namespace here — doing so + // is what previously pinned them on disk forever. + known := 0 + for chunk := range reported { + if info, ok := ms.Chunks[chunk]; ok { + info.Replicas[address] = true + known++ } } - logger.Info("Chunk server registered", "address", req.GetServerAddress()) + + logger.Info("Chunk server registered", + "address", address, "reported_chunks", len(reported), "known_chunks", known) return &masterpb.RegisterChunkServerResponse{Success: true}, nil } func (ms *MasterServer) Heartbeat(stream grpc.BidiStreamingServer[masterpb.HeartbeatRequest, masterpb.HeartbeatResponse]) error { - // 3 types - // 1st is master knows --> but the task is underreplicated - // master knows and the task is overreplicated - // 2nd and 3rd are handled already - for { msg, err := stream.Recv() if err == io.EOF { - break + return nil } if err != nil { return err } - logger.Debug("Heartbeat received", "server", msg.GetServerAddress()) - // Check if the server address exists + serverAddress := msg.GetServerAddress() + logger.Debug("Heartbeat received", "server", serverAddress) ms.mu.Lock() - - if _, ok := ms.ChunkServers[serverAddress]; !ok { + info, registered := ms.ChunkServers[serverAddress] + if !registered { ms.mu.Unlock() - return fmt.Errorf("server not registered") - } - ms.ChunkServers[serverAddress].FreeStorage = msg.GetFreeStorage() - ms.ChunkServers[serverAddress].LastHeartbeat = time.Now().UTC() - - serverChunks := make(map[string]bool) - - for _, chunk := range msg.GetChunks() { - serverChunks[chunk] = true + return status.Errorf(codes.FailedPrecondition, "chunkserver %s is not registered", serverAddress) } + info.FreeStorage = msg.GetFreeStorage() + info.LastHeartbeat = time.Now().UTC() - // find missing chunks - missingChunks := make([]string, 0) - del := make(map[string]*masterpb.DeleteTask, 0) - rep := make([]string, 0) - for chunk := range ms.ChunkServers[serverAddress].Chunks { - - if _, ok := serverChunks[chunk]; !ok { - missingChunks = append(missingChunks, chunk) - } else { - // if the master knows the task and it is present - // check the replication factor - if chunkInfo, ok := ms.Chunks[chunk]; ok { - if len(chunkInfo.Replicas) > REPLICATION_FACTOR { - del[chunk] = &masterpb.DeleteTask{ChunkId: chunk} - } - if len(chunkInfo.Replicas) < REPLICATION_FACTOR { - rep = append(rep, chunk) - } - } - } - - } - - newChunks := make([]string, 0) - - for chunk := range serverChunks { - if _, ok := ms.ChunkServers[serverAddress].Chunks[chunk]; !ok { - newChunks = append(newChunks, chunk) - } - } - - rep1 := ms.processMissingChunks(serverAddress, missingChunks) - // delete task is triggered 3 times - rep2, deleteTask := ms.processNewChunks(serverAddress, newChunks) - - rep1 = append(rep1, rep2...) - rep1 = append(rep1, rep...) - for _, task := range deleteTask { - del[task.GetChunkId()] = &masterpb.DeleteTask{ChunkId: task.ChunkId} - } - ms.buildReplicationTasks(rep1) - replicationTask := make([]*masterpb.ReplicationTask, 0) - for _, work := range ms.ReplicationWorks[serverAddress] { - replicationTask = append(replicationTask, &masterpb.ReplicationTask{ - ChunkId: work.ChunkID, - TargetAddress: work.TargetAddress, - }) - } - deleteTask = deleteTask[:0] + replicationTasks, deleteTasks := ms.reconcile(serverAddress, msg.GetChunks()) + ms.mu.Unlock() - for _, val := range del { - deleteTask = append(deleteTask, val) - } - // clear assigned tasks - ms.ChunkServers[serverAddress].Chunks = serverChunks - logger.Debug("Heartbeat response", "server", msg.GetServerAddress(), "replication_tasks", len(replicationTask), "delete_tasks", len(deleteTask)) + // Send outside the lock. stream.Send blocks when the chunkserver stops + // reading, and holding the global mutex across it would let one slow + // chunkserver stall every other RPC in the cluster. + logger.Debug("Heartbeat response", + "server", serverAddress, + "replication_tasks", len(replicationTasks), + "delete_tasks", len(deleteTasks)) if err := stream.Send(&masterpb.HeartbeatResponse{ - ReplicationTasks: replicationTask, - DeleteTasks: deleteTask, + ReplicationTasks: replicationTasks, + DeleteTasks: deleteTasks, }); err != nil { return err } - ms.ReplicationWorks[serverAddress] = make([]*ReplicationWork, 0) - for _, task := range deleteTask { // positive assumption - if chunkInfo, ok := ms.Chunks[task.GetChunkId()]; ok { - delete(chunkInfo.Replicas, serverAddress) - } - } - ms.mu.Unlock() } - return nil - } -func (ms *MasterServer) processMissingChunks(serverAddress string, missingChunks []string) []string { +// reconcile brings the master's view of one chunkserver in line with what that +// server just reported, and returns the work to hand back to it. +// +// The chunkserver's report is authoritative for its own contents: anything it +// lists, it has; anything it omits, it no longer has. Deriving state this way +// (rather than optimistically assuming previously dispatched tasks succeeded) +// means a dropped or failed task simply shows up again on the next heartbeat. +// +// Must be called with ms.mu held. +func (ms *MasterServer) reconcile(serverAddress string, reportedChunks []string) ([]*masterpb.ReplicationTask, []*masterpb.DeleteTask) { + info := ms.ChunkServers[serverAddress] + + reported := make(map[string]bool, len(reportedChunks)) + for _, chunk := range reportedChunks { + reported[chunk] = true + } - rep := make([]string, 0) - for _, chunk := range missingChunks { - chunkInfo, ok := ms.Chunks[chunk] - if !ok { + // Chunks we believed this server held but which it no longer reports. + underReplicated := make(map[string]bool) + for chunk := range info.Chunks { + if reported[chunk] { continue } - delete(chunkInfo.Replicas, serverAddress) + if chunkInfo, ok := ms.Chunks[chunk]; ok { + delete(chunkInfo.Replicas, serverAddress) + if len(chunkInfo.Replicas) < ms.cfg.ReplicationFactor { + underReplicated[chunk] = true + } + } + } - // if no file -> skip replication - if chunkInfo.FileName == "" || ms.Files[chunkInfo.FileName] == nil { - if len(chunkInfo.Replicas) == 0 { - delete(ms.Chunks, chunk) + deleteTasks := make([]*masterpb.DeleteTask, 0) + collectOrphans := time.Since(ms.bootTime) > ms.cfg.OrphanGracePeriod + + for chunk := range reported { + chunkInfo, known := ms.Chunks[chunk] + if !known { + // No namespace entry: this chunk belongs to no file. + if collectOrphans { + logger.Warn("Reclaiming orphaned chunk", "chunk", chunk, "server", serverAddress) + deleteTasks = append(deleteTasks, &masterpb.DeleteTask{ChunkId: chunk}) } continue } + chunkInfo.Replicas[serverAddress] = true + } - // if no replicas left -> metadata becomes invalid - if len(chunkInfo.Replicas) == 0 { - delete(ms.Chunks, chunk) + // Now that replica sets are accurate, decide over- and under-replication. + for chunk := range reported { + chunkInfo, known := ms.Chunks[chunk] + if !known { continue } - if len(chunkInfo.Replicas) < REPLICATION_FACTOR { - rep = append(rep, chunk) + switch { + case len(chunkInfo.Replicas) > ms.cfg.ReplicationFactor: + // Pick victims deterministically so that independent heartbeats agree + // on which replicas to drop. Without this, every server holding the + // chunk can be told to delete its copy and the data is lost. + if ms.isExcessReplica(chunkInfo, serverAddress) { + deleteTasks = append(deleteTasks, &masterpb.DeleteTask{ChunkId: chunk}) + } + case len(chunkInfo.Replicas) < ms.cfg.ReplicationFactor: + underReplicated[chunk] = true } } - return rep -} - -func (ms *MasterServer) processNewChunks(serverAddress string, newChunks []string) ([]string, []*masterpb.DeleteTask) { - rep, deleteTask := make([]string, 0), make([]*masterpb.DeleteTask, 0) + info.Chunks = reported - for _, chunk := range newChunks { - chunkInfo, ok := ms.Chunks[chunk] - if !ok { - deleteTask = append(deleteTask, &masterpb.DeleteTask{ChunkId: chunk}) - continue + for chunk := range underReplicated { + if chunkInfo, ok := ms.Chunks[chunk]; ok && len(chunkInfo.Replicas) == 0 { + // Metadata is kept: the namespace is durable, so we can still describe + // the file's structure and heal it if a replica comes back. + logger.Error("Chunk has no reachable replicas", "chunk", chunk, "file", chunkInfo.FileName) } + } + ms.buildReplicationTasks(keys(underReplicated)) - if chunkInfo.FileName == "" || ms.Files[chunkInfo.FileName] == nil { - deleteTask = append(deleteTask, &masterpb.DeleteTask{ChunkId: chunk}) - continue - } + replicationTasks := make([]*masterpb.ReplicationTask, 0, len(ms.ReplicationWorks[serverAddress])) + for _, work := range ms.ReplicationWorks[serverAddress] { + replicationTasks = append(replicationTasks, &masterpb.ReplicationTask{ + ChunkId: work.ChunkID, + TargetAddress: work.TargetAddress, + }) + } + delete(ms.ReplicationWorks, serverAddress) - chunkInfo.Replicas[serverAddress] = true + return replicationTasks, deleteTasks +} - if len(chunkInfo.Replicas) > REPLICATION_FACTOR { - deleteTask = append(deleteTask, &masterpb.DeleteTask{ChunkId: chunk}) - continue - } - if len(chunkInfo.Replicas) < REPLICATION_FACTOR { - rep = append(rep, chunk) +// isExcessReplica reports whether serverAddress is one of the replicas to drop +// when a chunk is over-replicated. Replicas sort by address and the first +// ReplicationFactor entries are kept, so every heartbeat reaches the same answer. +func (ms *MasterServer) isExcessReplica(chunkInfo *ChunkInfo, serverAddress string) bool { + replicas := keys(chunkInfo.Replicas) + sort.Strings(replicas) + for i, addr := range replicas { + if addr == serverAddress { + return i >= ms.cfg.ReplicationFactor } + } + return false +} +func keys(set map[string]bool) []string { + out := make([]string, 0, len(set)) + for k := range set { + out = append(out, k) } - return rep, deleteTask + return out +} +func (ms *MasterServer) isAlive(info *ChunkServerInfo) bool { + return time.Since(info.LastHeartbeat) <= ms.cfg.LiveThreshold } -// buildReplicationTasks MUST be called with ms.mu held +// buildReplicationTasks queues copies for under-replicated chunks. +// Must be called with ms.mu held. func (ms *MasterServer) buildReplicationTasks(rep []string) { - - // local snapshot of free server storage - serverStorage := make(map[string]int64) - for addr, info := range ms.ChunkServers { - serverStorage[addr] = info.FreeStorage - } + // Local view of free storage so one pass does not over-commit a server. + serverStorage := make(map[string]int64, len(ms.ChunkServers)) type srv struct { addr string free int64 alive bool } servers := make([]srv, 0, len(ms.ChunkServers)) - for addr, info := range ms.ChunkServers { - servers = append(servers, srv{ - addr: addr, - free: info.FreeStorage, - alive: time.Since(info.LastHeartbeat) <= LIVE_THRESHOLD, - }) + serverStorage[addr] = info.FreeStorage + servers = append(servers, srv{addr: addr, free: info.FreeStorage, alive: ms.isAlive(info)}) } - // highest free storage first + // Highest free storage first, address as tiebreak for deterministic placement. sort.Slice(servers, func(i, j int) bool { - return servers[i].free > servers[j].free + if servers[i].free != servers[j].free { + return servers[i].free > servers[j].free + } + return servers[i].addr < servers[j].addr }) - for _, chunkID := range rep { + // Chunks already queued for a given target, so repeated passes before the + // first copy lands do not pile up duplicate work. + pending := make(map[string]bool) + for _, works := range ms.ReplicationWorks { + for _, w := range works { + pending[w.ChunkID+"@"+w.TargetAddress] = true + } + } - ci, ok := ms.Chunks[chunkID] + sort.Strings(rep) + for _, chunkID := range rep { + chunkInfo, ok := ms.Chunks[chunkID] if !ok { continue } - - replicasNeeded := REPLICATION_FACTOR - len(ci.Replicas) + replicasNeeded := ms.cfg.ReplicationFactor - len(chunkInfo.Replicas) if replicasNeeded <= 0 { continue } - // source selection + + // A live replica has to supply the data. var source string - for replica := range ci.Replicas { - if info := ms.ChunkServers[replica]; time.Since(info.LastHeartbeat) <= LIVE_THRESHOLD { + sources := keys(chunkInfo.Replicas) + sort.Strings(sources) + for _, replica := range sources { + if info, ok := ms.ChunkServers[replica]; ok && ms.isAlive(info) { source = replica break } } if source == "" { - continue // no alive replica to supply the chunk + continue } - // target selection - targets := make([]string, 0) for _, s := range servers { - if replicasNeeded == 0 { break } - - if !s.alive { + if !s.alive || chunkInfo.Replicas[s.addr] { continue } - - if _, alreadyReplica := ci.Replicas[s.addr]; alreadyReplica { + if serverStorage[s.addr] < CHUNK_SIZE_MB { continue } - - if serverStorage[s.addr] < CHUNK_SIZE_MB { + key := chunkID + "@" + s.addr + if pending[key] { continue } - - targets = append(targets, s.addr) - serverStorage[s.addr] -= CHUNK_SIZE_MB // simulation + pending[key] = true + serverStorage[s.addr] -= CHUNK_SIZE_MB replicasNeeded-- - } - // add tasks to ReplicationWorks - for _, dest := range targets { - work := &ReplicationWork{ - TargetAddress: dest, + ms.ReplicationWorks[source] = append(ms.ReplicationWorks[source], &ReplicationWork{ + TargetAddress: s.addr, ChunkID: chunkID, - } - ms.ReplicationWorks[source] = append(ms.ReplicationWorks[source], work) + }) } } } -func (ms *MasterServer) GetFileInfo(ctx context.Context, fileInfo *masterpb.GetFileInfoRequest) (*masterpb.GetFileInfoResponse, error) { - +func (ms *MasterServer) GetFileInfo(ctx context.Context, req *masterpb.GetFileInfoRequest) (*masterpb.GetFileInfoResponse, error) { ms.mu.Lock() defer ms.mu.Unlock() - fileMetaData, exists := ms.Files[fileInfo.GetFileName()] + fileMetaData, exists := ms.Files[req.GetFileName()] if !exists { - return &masterpb.GetFileInfoResponse{ - FileInfo: make([]*masterpb.FileChunkInfo, 0), - }, fmt.Errorf("file does not exist") + return nil, status.Errorf(codes.NotFound, "file %q does not exist", req.GetFileName()) } - chunks := make([]*masterpb.FileChunkInfo, 0) - + // Chunks are returned in allocation order, so the client does not have to + // recover ordering by parsing chunk IDs. + chunks := make([]*masterpb.FileChunkInfo, 0, len(fileMetaData.Chunks)) for _, chunkID := range fileMetaData.Chunks { chunk, exists := ms.Chunks[chunkID] if !exists { - continue + return nil, status.Errorf(codes.Internal, "file %q references unknown chunk %q", req.GetFileName(), chunkID) } - replicas := make([]string, 0) - for replica := range chunk.Replicas { - replicas = append(replicas, replica) + replicas := keys(chunk.Replicas) + sort.Strings(replicas) + if len(replicas) == 0 { + return nil, status.Errorf(codes.Unavailable, "chunk %q has no available replicas", chunkID) } - chunks = append(chunks, &masterpb.FileChunkInfo{ ChunkId: chunkID, ReplicaServers: replicas, }) } - return &masterpb.GetFileInfoResponse{ - FileInfo: chunks, - }, nil + return &masterpb.GetFileInfoResponse{FileInfo: chunks}, nil } func (ms *MasterServer) AllocateChunk(ctx context.Context, request *masterpb.AllocateChunkRequest) (*masterpb.AllocateChunkResponse, error) { - - // gets the file name and chunk index number as input - // send the chunkId and the replica servers fileName := request.GetFileName() - chunkIdx := request.GetChunkIndex() + if fileName == "" { + return nil, status.Error(codes.InvalidArgument, "file_name is required") + } + chunkID := fileName + "_" + strconv.Itoa(int(request.GetChunkIndex())) ms.mu.Lock() defer ms.mu.Unlock() - fileMetaData, exists := ms.Files[fileName] - if !exists { - fileMetaData = &FileMetaData{ - FileName: fileName, - Chunks: make([]string, 0), - } - ms.Files[fileName] = fileMetaData - } - - chunkID := fileName + "_" + strconv.Itoa(int(chunkIdx)) if _, exists := ms.Chunks[chunkID]; exists { - return &masterpb.AllocateChunkResponse{ - ChunkId: chunkID, - ReplicaServers: make([]string, 0), - }, fmt.Errorf(" chunk already exists ") + return nil, status.Errorf(codes.AlreadyExists, "chunk %q already exists", chunkID) } - // alive servers - - validServers := make([]*ChunkServerInfo, 0) + validServers := make([]*ChunkServerInfo, 0, len(ms.ChunkServers)) for _, serverInfo := range ms.ChunkServers { - if time.Since(serverInfo.LastHeartbeat) > LIVE_THRESHOLD || serverInfo.FreeStorage < CHUNK_SIZE_MB { + if !ms.isAlive(serverInfo) || serverInfo.FreeStorage < CHUNK_SIZE_MB { continue } validServers = append(validServers, serverInfo) } - - if len(validServers) < REPLICATION_FACTOR { - return &masterpb.AllocateChunkResponse{ - ChunkId: chunkID, - ReplicaServers: make([]string, 0), - }, fmt.Errorf(" server count criteria not met ") + if len(validServers) < ms.cfg.ReplicationFactor { + return nil, status.Errorf(codes.ResourceExhausted, + "need %d live chunkservers with free space, have %d", ms.cfg.ReplicationFactor, len(validServers)) } sort.Slice(validServers, func(i, j int) bool { - return validServers[i].FreeStorage > validServers[j].FreeStorage + if validServers[i].FreeStorage != validServers[j].FreeStorage { + return validServers[i].FreeStorage > validServers[j].FreeStorage + } + return validServers[i].Address < validServers[j].Address }) - replicaServers := []string{} - replicaServerMap := make(map[string]bool) + // Commit the namespace change before handing the client somewhere to write. + // If this fails the client gets an error and no chunk is created, which is + // recoverable; the reverse would leave data the master forgets about. + if err := ms.store.AppendAllocateChunk(fileName, chunkID); err != nil { + logger.Error("Namespace log append failed", "file", fileName, "chunk", chunkID, "error", err) + return nil, status.Errorf(codes.Internal, "failed to persist namespace: %v", err) + } + + fileMetaData, exists := ms.Files[fileName] + if !exists { + fileMetaData = &FileMetaData{FileName: fileName, Chunks: make([]string, 0, 1)} + ms.Files[fileName] = fileMetaData + } - for i := 0; i < REPLICATION_FACTOR; i++ { - replicaServers = append(replicaServers, validServers[i].Address) - replicaServerMap[replicaServers[i]] = true - ms.ChunkServers[replicaServers[i]].Chunks[chunkID] = true - ms.ChunkServers[replicaServers[i]].FreeStorage -= CHUNK_SIZE_MB + replicaServers := make([]string, 0, ms.cfg.ReplicationFactor) + replicaServerMap := make(map[string]bool, ms.cfg.ReplicationFactor) + for i := 0; i < ms.cfg.ReplicationFactor; i++ { + addr := validServers[i].Address + replicaServers = append(replicaServers, addr) + replicaServerMap[addr] = true + validServers[i].Chunks[chunkID] = true + validServers[i].FreeStorage -= CHUNK_SIZE_MB } fileMetaData.Chunks = append(fileMetaData.Chunks, chunkID) @@ -471,51 +565,62 @@ func (ms *MasterServer) AllocateChunk(ctx context.Context, request *masterpb.All }, nil } -// Here the master assumes that the chunk might come back online -// What should be the time , when i should consider it dead ?? -// if it exceeds more than 30 secs --> the server is dead -// otherwise it is just offline for now -// current implementation calls the register cs rpc +// detectDeadChunkServer drops chunkservers that have stopped heartbeating and +// queues replacement copies for the chunks they held. func (ms *MasterServer) detectDeadChunkServer() { - // remove it from the cache - - // problems while implementing --> need to also keep in mind that there were a few chunks that were present - // in the dead server - - // first need to replicate those chunks - // here the system assumes that the chunk server might come back online ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + for range ticker.C { - deadServers := make([]string, 0) - chunks := make([]string, 0) ms.mu.Lock() - for serverAddress, info := range ms.ChunkServers { + deadServers := make([]string, 0) + affected := make(map[string]bool) - if time.Since(info.LastHeartbeat) > LIVE_THRESHOLD { // live threshold - // server is dead ( assumption --> so create replication tasks ) - // create a replication slice and build replication task - // i only need the file chunks right --> yes !! - for chunk, _ := range ms.ChunkServers[serverAddress].Chunks { - chunks = append(chunks, chunk) - if chunkInfo, ok := ms.Chunks[chunk]; ok { - delete(chunkInfo.Replicas, serverAddress) - } + for serverAddress, info := range ms.ChunkServers { + if ms.isAlive(info) { + continue + } + for chunk := range info.Chunks { + if chunkInfo, ok := ms.Chunks[chunk]; ok { + delete(chunkInfo.Replicas, serverAddress) + affected[chunk] = true } - deadServers = append(deadServers, serverAddress) } - + deadServers = append(deadServers, serverAddress) } - // removes the chunk server from master server for _, dead := range deadServers { logger.Warn("Dead chunk server detected", "address", dead) delete(ms.ChunkServers, dead) + delete(ms.ReplicationWorks, dead) + } + if len(affected) > 0 { + ms.buildReplicationTasks(keys(affected)) + } + ms.mu.Unlock() + } +} + +// checkpointLoop folds the namespace log into a checkpoint so recovery stays fast. +func (ms *MasterServer) checkpointLoop() { + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + + for range ticker.C { + if !ms.store.CheckpointDue() { + continue } - if len(deadServers) > 0 { - ms.buildReplicationTasks(chunks) + ms.mu.Lock() + files := make(map[string][]string, len(ms.Files)) + for name, meta := range ms.Files { + files[name] = append([]string(nil), meta.Chunks...) } ms.mu.Unlock() - // time.Sleep(time.Second * 30) + if err := ms.store.Checkpoint(files); err != nil { + logger.Error("Checkpoint failed", "error", err) + continue + } + logger.Info("Namespace checkpoint written", "files", len(files)) } } diff --git a/internal/master/server_test.go b/internal/master/server_test.go new file mode 100644 index 0000000..22713e6 --- /dev/null +++ b/internal/master/server_test.go @@ -0,0 +1,322 @@ +package master + +import ( + "context" + "sort" + "strconv" + "testing" + "time" + + "dfs/dfs/masterpb" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func testConfig(t *testing.T) Config { + t.Helper() + cfg := DefaultConfig() + cfg.MetadataDir = t.TempDir() + cfg.ReplicationFactor = 2 + return cfg +} + +func newTestMaster(t *testing.T, cfg Config) *MasterServer { + t.Helper() + ms, err := NewMasterServer(cfg) + if err != nil { + t.Fatalf("NewMasterServer: %v", err) + } + t.Cleanup(func() { ms.Close() }) + return ms +} + +// register adds a live chunkserver with plenty of free space. +func register(t *testing.T, ms *MasterServer, addr string, chunks ...string) { + t.Helper() + _, err := ms.RegisterChunkServer(context.Background(), &masterpb.RegisterChunkServerRequest{ + ServerAddress: addr, + FreeStorage: 1_000_000, + Chunks: chunks, + }) + if err != nil { + t.Fatalf("RegisterChunkServer(%s): %v", addr, err) + } +} + +func allocate(t *testing.T, ms *MasterServer, file string, idx int32) *masterpb.AllocateChunkResponse { + t.Helper() + resp, err := ms.AllocateChunk(context.Background(), &masterpb.AllocateChunkRequest{ + FileName: file, ChunkIndex: idx, + }) + if err != nil { + t.Fatalf("AllocateChunk(%s, %d): %v", file, idx, err) + } + return resp +} + +func codeOf(err error) codes.Code { + return status.Code(err) +} + +func TestAllocateChunkReturnsReplicationFactorReplicas(t *testing.T) { + ms := newTestMaster(t, testConfig(t)) + register(t, ms, "cs1") + register(t, ms, "cs2") + register(t, ms, "cs3") + + resp := allocate(t, ms, "f.bin", 0) + if got := len(resp.GetReplicaServers()); got != 2 { + t.Fatalf("got %d replicas, want 2", got) + } + if resp.GetChunkId() != "f.bin_0" { + t.Fatalf("chunk id = %q", resp.GetChunkId()) + } +} + +func TestAllocateChunkFailsWithoutEnoughServers(t *testing.T) { + ms := newTestMaster(t, testConfig(t)) + register(t, ms, "cs1") + + _, err := ms.AllocateChunk(context.Background(), &masterpb.AllocateChunkRequest{FileName: "f.bin"}) + if codeOf(err) != codes.ResourceExhausted { + t.Fatalf("got %v (%v), want ResourceExhausted", err, codeOf(err)) + } +} + +func TestAllocateChunkRejectsDuplicate(t *testing.T) { + ms := newTestMaster(t, testConfig(t)) + register(t, ms, "cs1") + register(t, ms, "cs2") + allocate(t, ms, "f.bin", 0) + + _, err := ms.AllocateChunk(context.Background(), &masterpb.AllocateChunkRequest{FileName: "f.bin", ChunkIndex: 0}) + if codeOf(err) != codes.AlreadyExists { + t.Fatalf("got %v (%v), want AlreadyExists", err, codeOf(err)) + } +} + +func TestGetFileInfoUnknownFile(t *testing.T) { + ms := newTestMaster(t, testConfig(t)) + _, err := ms.GetFileInfo(context.Background(), &masterpb.GetFileInfoRequest{FileName: "nope"}) + if codeOf(err) != codes.NotFound { + t.Fatalf("got %v (%v), want NotFound", err, codeOf(err)) + } +} + +// Chunks must come back in allocation order, not sorted by ID: "f_10" sorts +// before "f_2" as a string, and reassembling in that order corrupts the file. +func TestGetFileInfoReturnsChunksInAllocationOrder(t *testing.T) { + ms := newTestMaster(t, testConfig(t)) + register(t, ms, "cs1") + register(t, ms, "cs2") + for i := int32(0); i < 12; i++ { + allocate(t, ms, "f", i) + } + + resp, err := ms.GetFileInfo(context.Background(), &masterpb.GetFileInfoRequest{FileName: "f"}) + if err != nil { + t.Fatalf("GetFileInfo: %v", err) + } + for i, info := range resp.GetFileInfo() { + want := "f_" + strconv.Itoa(i) + if info.GetChunkId() != want { + t.Fatalf("chunk %d = %q, want %q", i, info.GetChunkId(), want) + } + } +} + +// The regression that made master restarts lose every file: metadata lived only +// in memory, so a restart left chunks stranded on disk and unreachable. +func TestNamespaceSurvivesRestart(t *testing.T) { + cfg := testConfig(t) + + first := newTestMaster(t, cfg) + register(t, first, "cs1") + register(t, first, "cs2") + allocate(t, first, "f.bin", 0) + allocate(t, first, "f.bin", 1) + first.Close() + + // A fresh master over the same metadata directory, as after a crash. + second := newTestMaster(t, cfg) + // Chunkservers re-register and re-announce what they hold. + register(t, second, "cs1", "f.bin_0", "f.bin_1") + register(t, second, "cs2", "f.bin_0", "f.bin_1") + + resp, err := second.GetFileInfo(context.Background(), &masterpb.GetFileInfoRequest{FileName: "f.bin"}) + if err != nil { + t.Fatalf("GetFileInfo after restart: %v", err) + } + if got := len(resp.GetFileInfo()); got != 2 { + t.Fatalf("got %d chunks after restart, want 2", got) + } + for _, info := range resp.GetFileInfo() { + replicas := info.GetReplicaServers() + sort.Strings(replicas) + if len(replicas) != 2 { + t.Fatalf("chunk %s has replicas %v, want both servers relearned from registration", + info.GetChunkId(), replicas) + } + } +} + +// Replica locations are soft state: they must come back from chunkservers, not +// from the on-disk namespace. +func TestReplicaLocationsAreNotPersisted(t *testing.T) { + cfg := testConfig(t) + + first := newTestMaster(t, cfg) + register(t, first, "cs1") + register(t, first, "cs2") + allocate(t, first, "f.bin", 0) + first.Close() + + second := newTestMaster(t, cfg) + // No chunkserver has registered yet, so the master must admit it does not + // know where the data is rather than serving a stale location. + _, err := second.GetFileInfo(context.Background(), &masterpb.GetFileInfoRequest{FileName: "f.bin"}) + if codeOf(err) != codes.Unavailable { + t.Fatalf("got %v (%v), want Unavailable before any chunkserver registers", err, codeOf(err)) + } +} + +func TestReconcileRecordsAndDropsReplicas(t *testing.T) { + ms := newTestMaster(t, testConfig(t)) + register(t, ms, "cs1") + register(t, ms, "cs2") + allocate(t, ms, "f.bin", 0) + + // cs1 reports it no longer holds the chunk. + ms.mu.Lock() + ms.reconcile("cs1", nil) + replicas := len(ms.Chunks["f.bin_0"].Replicas) + ms.mu.Unlock() + + if replicas != 1 { + t.Fatalf("replica count = %d, want 1 after cs1 reported it dropped the chunk", replicas) + } + + // It reappears; the location must be relearned. + ms.mu.Lock() + ms.reconcile("cs1", []string{"f.bin_0"}) + replicas = len(ms.Chunks["f.bin_0"].Replicas) + ms.mu.Unlock() + + if replicas != 2 { + t.Fatalf("replica count = %d, want 2 after cs1 reported the chunk again", replicas) + } +} + +func TestReconcileQueuesReplicationWhenUnderReplicated(t *testing.T) { + ms := newTestMaster(t, testConfig(t)) + register(t, ms, "cs1") + register(t, ms, "cs2") + register(t, ms, "cs3") + allocate(t, ms, "f.bin", 0) + + ms.mu.Lock() + // cs1 loses its copy, leaving one replica against a factor of two. + ms.reconcile("cs1", nil) + queued := 0 + for _, works := range ms.ReplicationWorks { + queued += len(works) + } + ms.mu.Unlock() + + if queued != 1 { + t.Fatalf("queued %d replication tasks, want 1", queued) + } +} + +// With three replicas and a factor of two, exactly one server may be told to +// delete. The previous logic could tell every holder to delete its copy. +func TestOverReplicationDeletesExactlyOneReplica(t *testing.T) { + ms := newTestMaster(t, testConfig(t)) + register(t, ms, "cs1") + register(t, ms, "cs2") + allocate(t, ms, "f.bin", 0) + register(t, ms, "cs3", "f.bin_0") + + deletes := map[string]int{} + ms.mu.Lock() + for _, addr := range []string{"cs1", "cs2", "cs3"} { + _, del := ms.reconcile(addr, []string{"f.bin_0"}) + deletes[addr] = len(del) + } + total := deletes["cs1"] + deletes["cs2"] + deletes["cs3"] + ms.mu.Unlock() + + if total != 1 { + t.Fatalf("total delete tasks = %d (%v), want exactly 1", total, deletes) + } +} + +// Orphans are only collected after the grace period, otherwise a restarting +// master would delete live data before its chunkservers had re-registered. +func TestOrphanCollectionRespectsGracePeriod(t *testing.T) { + cfg := testConfig(t) + cfg.OrphanGracePeriod = time.Hour + ms := newTestMaster(t, cfg) + register(t, ms, "cs1", "stray_0") + + ms.mu.Lock() + _, del := ms.reconcile("cs1", []string{"stray_0"}) + ms.mu.Unlock() + if len(del) != 0 { + t.Fatalf("got %d delete tasks during the grace period, want 0", len(del)) + } + + // Simulate the grace period having elapsed. + ms.mu.Lock() + ms.bootTime = time.Now().Add(-2 * time.Hour) + _, del = ms.reconcile("cs1", []string{"stray_0"}) + ms.mu.Unlock() + if len(del) != 1 { + t.Fatalf("got %d delete tasks after the grace period, want 1", len(del)) + } +} + +// Registration must not invent namespace entries for chunks it has never heard +// of; doing so is what pinned orphaned chunks on disk forever. +func TestRegisterDoesNotResurrectUnknownChunks(t *testing.T) { + ms := newTestMaster(t, testConfig(t)) + register(t, ms, "cs1", "ghost_0") + + ms.mu.Lock() + _, known := ms.Chunks["ghost_0"] + ms.mu.Unlock() + + if known { + t.Fatal("registration created namespace metadata for an unknown chunk") + } +} + +// A server that stops heartbeating loses its replicas and its chunks are queued +// for re-replication elsewhere. +func TestDeadServerReplicasAreDropped(t *testing.T) { + cfg := testConfig(t) + cfg.LiveThreshold = time.Millisecond + ms := newTestMaster(t, cfg) + + register(t, ms, "cs1") + register(t, ms, "cs2") + allocate(t, ms, "f.bin", 0) + + time.Sleep(5 * time.Millisecond) // both servers are now past the threshold + + ms.mu.Lock() + for _, info := range ms.ChunkServers { + if ms.isAlive(info) { + t.Fatal("servers should be considered dead") + } + } + ms.mu.Unlock() + + // A live server arriving cannot receive copies until a source is alive again, + // so allocation must fail rather than silently under-replicate. + _, err := ms.AllocateChunk(context.Background(), &masterpb.AllocateChunkRequest{FileName: "g.bin"}) + if codeOf(err) != codes.ResourceExhausted { + t.Fatalf("got %v (%v), want ResourceExhausted with no live servers", err, codeOf(err)) + } +} diff --git a/internal/test/testChunker.go b/internal/test/testChunker.go deleted file mode 100644 index e89678a..0000000 --- a/internal/test/testChunker.go +++ /dev/null @@ -1,36 +0,0 @@ -package main - -import ( - "bytes" - "dfs/internal/client/uploader" - "fmt" - "io" -) - -func main() { - - chunker, err := uploader.NewChunker("./output.txt", int64(16)) - if err != nil { - fmt.Println(err) - } - buff := make([]byte, 16) - for { - data, _, err := chunker.NextChunk(buff) - if err == io.EOF { - break - } - if err != nil { - fmt.Println(err) - } - fmt.Println(bytes.NewBuffer(data).String()) - fmt.Println("*****************") - } - fmt.Println("Seek") - data, _ := chunker.GetChunkAtIndex(0, buff) - fmt.Println(bytes.NewBuffer(data).String()) - - fmt.Println("Next") - data, _, _ = chunker.NextChunk(buff) - fmt.Println(bytes.NewBuffer(data).String()) - chunker.Close() -}