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() -}