diff --git a/internal/server/server.go b/internal/server/server.go index b0588b2..088ca87 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -1,24 +1,35 @@ package server import ( + "bytes" "context" "encoding/json" "fmt" "io" "net/http" + "strconv" + "strings" "sync/atomic" "time" "github.com/amirderis/DHT/internal/config" + "github.com/amirderis/DHT/internal/ring" "github.com/amirderis/DHT/internal/storage" "github.com/amirderis/DHT/pkg/api" ) +const ( + readConsistencyHeader = "X-Consistency-R" + writeConsistencyHeader = "X-Consistency-W" +) + type HTTPServer struct { cfg *config.Config server *http.Server readyFlag atomic.Bool storage storage.Engine + ring *ring.Ring + client *http.Client } func NewHTTPServer(cfg *config.Config) *HTTPServer { @@ -26,8 +37,15 @@ func NewHTTPServer(cfg *config.Config) *HTTPServer { s := &HTTPServer{ cfg: cfg, storage: storage.NewInMemory(), + ring: ring.New(20), // 20 virtual nodes per physical node + client: &http.Client{ + Timeout: 5 * time.Second, + }, } + // Initialize ring with this node + s.ring.AddNode(ring.NodeID(cfg.NodeID), cfg.BindAddr) + // Health and readiness endpoints mux.HandleFunc("/healthz", s.handleHealth) mux.HandleFunc("/readyz", s.handleReady) @@ -35,6 +53,9 @@ func NewHTTPServer(cfg *config.Config) *HTTPServer { // KV API endpoints mux.HandleFunc("/kv/", s.handleKV) + // Internal storage endpoints + mux.HandleFunc("/internal/storage/", s.handleInternalStorage) + s.server = &http.Server{ Addr: cfg.BindAddr, Handler: mux, @@ -72,7 +93,7 @@ func (s *HTTPServer) handleReady(w http.ResponseWriter, r *http.Request) { _, _ = fmt.Fprintln(w, "ready") } -// handleKV routes GET/PUT/DELETE requests to appropriate handlers +// handleKV routes GET/PUT/DELETE requests for a key to appropriate handlers func (s *HTTPServer) handleKV(w http.ResponseWriter, r *http.Request) { key := r.URL.Path[len("/kv/"):] if key == "" { @@ -88,29 +109,63 @@ func (s *HTTPServer) handleKV(w http.ResponseWriter, r *http.Request) { case http.MethodDelete: s.handleDelete(w, r, key) default: - s.writeError(w, http.StatusMethodNotAllowed, "method not allowed") + s.writeError(w, http.StatusMethodNotAllowed, "method not allowed: "+r.Method) } } -func (s *HTTPServer) handleGet(w http.ResponseWriter, _ *http.Request, key string) { - value, found := s.storage.Get(key) +func (s *HTTPServer) handleGet(w http.ResponseWriter, r *http.Request, key string) { + readQuorum := s.getQuorumFromHeader(r, readConsistencyHeader, s.cfg.ReadQuorum) + + preferenceList, err := s.ring.GetPreferenceList(key, s.cfg.ReplicationFactor) + if err != nil { + s.writeError(w, http.StatusInternalServerError, "failed to get preference list for key: "+key) + return + } + + // If we only have one node or read quorum=1, just read locally + if len(preferenceList) == 1 || readQuorum == 1 { + value, found := s.storage.Get(key) + response := api.GetResponse{ + Key: key, + Value: value, + Found: found, + } + if found { + w.WriteHeader(http.StatusOK) + } else { + w.WriteHeader(http.StatusNotFound) + } + s.writeJSON(w, response) + return + } - response := api.GetResponse{ - Key: key, - Value: value, - Found: found, + // Read from multiple nodes + responses := s.readFromNodes(key, preferenceList, readQuorum) + if len(responses) < readQuorum { + message := fmt.Sprintf("expected %d replicas, got %d", readQuorum, len(responses)) + s.writeError(w, http.StatusServiceUnavailable, message) + return } - if found { + // For now, return the first successful response + // TODO: Implement conflict resolution in Phase 3 + var response api.GetResponse + for _, resp := range responses { + if resp.Found { + response = resp + break + } + } + if response.Found { w.WriteHeader(http.StatusOK) } else { w.WriteHeader(http.StatusNotFound) } - s.writeJSON(w, response) } func (s *HTTPServer) handlePut(w http.ResponseWriter, r *http.Request, key string) { + writeQuorum := s.getQuorumFromHeader(r, writeConsistencyHeader, s.cfg.WriteQuorum) body, err := io.ReadAll(r.Body) if err != nil { s.writeError(w, http.StatusBadRequest, "failed to read request body") @@ -118,19 +173,106 @@ func (s *HTTPServer) handlePut(w http.ResponseWriter, r *http.Request, key strin } defer r.Body.Close() - if err := s.storage.Put(key, body); err != nil { - s.writeError(w, http.StatusInternalServerError, "failed to store value") + preferenceList, err := s.ring.GetPreferenceList(key, s.cfg.ReplicationFactor) + if err != nil { + s.writeError(w, http.StatusInternalServerError, "failed to get preference list for key: "+key) + return + } + + // Create version (placeholder for vector clock) + version := map[string]uint64{s.cfg.NodeID: 1} + + // If we only have one node or write quorum=1, just write locally + if len(preferenceList) == 1 || writeQuorum == 1 { + if err := s.storage.Put(key, body); err != nil { + s.writeError(w, http.StatusInternalServerError, "failed to store value") + return + } + + response := api.PutResponse{Version: version} + w.WriteHeader(http.StatusOK) + s.writeJSON(w, response) return } - response := api.PutResponse{ - Version: map[string]uint64{s.cfg.NodeID: 1}, // Placeholder for vector clock + // Write to multiple nodes + successCount := s.writeToNodes(key, body, version, preferenceList, writeQuorum) + if successCount < writeQuorum { + s.writeError(w, http.StatusServiceUnavailable, "insufficient replicas available for write quorum for key: "+key) + return } + response := api.PutResponse{Version: version} w.WriteHeader(http.StatusOK) s.writeJSON(w, response) } +// writeToNodes writes to multiple nodes and returns success count +func (s *HTTPServer) writeToNodes(key string, value []byte, version map[string]uint64, prefList []ring.NodeID, writeQuorum int) int { + successCount := 0 + + for _, nodeID := range prefList { + if successCount >= writeQuorum { + break + } + + // If it's this node, write locally + if nodeID == ring.NodeID(s.cfg.NodeID) { + if err := s.storage.Put(key, value); err == nil { + successCount++ + } else { + fmt.Printf("failed to write to local node %s for key: %s, error: %v\n", s.cfg.NodeID, key, err) + } + continue + } + + // Write to remote node + address, exists := s.ring.GetNodeAddress(nodeID) + if !exists { + fmt.Printf("node %s not found in ring for key: %s\n", nodeID, key) + continue + } + if err := s.writeToRemoteNode(address, key, value, version); err == nil { + successCount++ + } else { + fmt.Printf("failed to write to remote node %s for key: %s, error: %v\n", address, key, err) + } + } + return successCount +} + +func (s *HTTPServer) writeToRemoteNode(address, key string, value []byte, version map[string]uint64) error { + req := api.ReplicateRequest{ + Key: key, + Value: value, + Version: version, + } + var jsonData bytes.Buffer + if err := json.NewEncoder(&jsonData).Encode(req); err != nil { + return err + } + url := fmt.Sprintf("http://%s/internal/storage/%s", address, key) + resp, err := s.client.Post(url, "application/json", strings.NewReader(jsonData.String())) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("remote node %s returned status %d", address, resp.StatusCode) + } + + var result api.ReplicateResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return err + } + if !result.Success { + return fmt.Errorf("remote node %s failed to store value", address) + } + + return nil +} + func (s *HTTPServer) handleDelete(w http.ResponseWriter, _ *http.Request, key string) { if err := s.storage.Delete(key); err != nil { s.writeError(w, http.StatusInternalServerError, "failed to delete key") @@ -140,6 +282,51 @@ func (s *HTTPServer) handleDelete(w http.ResponseWriter, _ *http.Request, key st w.WriteHeader(http.StatusNoContent) } +func (s *HTTPServer) handleInternalStorage(w http.ResponseWriter, r *http.Request) { + key := r.URL.Path[len("/internal/storage/"):] + if key == "" { + s.writeError(w, http.StatusBadRequest, "key cannot be empty") + return + } + + switch r.Method { + case http.MethodGet: + value, found := s.storage.Get(key) + response := api.ReplicateGetResponse{ + Key: key, + Value: value, + Found: found, + } + if found { + w.WriteHeader(http.StatusOK) + } else { + w.WriteHeader(http.StatusNotFound) + } + s.writeJSON(w, response) + case http.MethodPost: + var req api.ReplicateRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + s.writeError(w, http.StatusBadRequest, "invalid request body") + return + } + if err := s.storage.Put(key, req.Value); err != nil { + response := api.ReplicateResponse{ + Success: false, + Error: "failed to store value", + } + w.WriteHeader(http.StatusInternalServerError) + s.writeJSON(w, response) + return + } + + response := api.ReplicateResponse{Success: true} + w.WriteHeader(http.StatusOK) + s.writeJSON(w, response) + default: + s.writeError(w, http.StatusMethodNotAllowed, "method not allowed: "+r.Method) + } +} + func (s *HTTPServer) writeJSON(w http.ResponseWriter, v any) { w.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(w).Encode(v); err != nil { @@ -154,3 +341,70 @@ func (s *HTTPServer) writeError(w http.ResponseWriter, status int, message strin errorResp := map[string]string{"error": message} json.NewEncoder(w).Encode(errorResp) } + +func (s *HTTPServer) getQuorumFromHeader(r *http.Request, headerName string, defaultValue int) int { + if headerValue := r.Header.Get(headerName); headerValue != "" { + var quorum int + quorum, err := strconv.Atoi(headerValue) + if err == nil && quorum > 0 { + return quorum + } + } + return defaultValue +} + +func (s *HTTPServer) readFromNodes(key string, prefList []ring.NodeID, readQuorum int) []api.GetResponse { + responses := make([]api.GetResponse, 0, len(prefList)) + + for _, nodeID := range prefList { + if len(responses) >= readQuorum { + break + } + + // If it's this node, read locally + if nodeID == ring.NodeID(s.cfg.NodeID) { + value, found := s.storage.Get(key) + responses = append(responses, api.GetResponse{ + Key: key, + Value: value, + Found: found, + }) + continue + } + + // Read from remote node + address, exists := s.ring.GetNodeAddress(nodeID) + if !exists { + continue + } + + resp, err := s.readFromRemoteNode(address, key) + if err == nil { + responses = append(responses, resp) + } + } + return responses +} + +func (s *HTTPServer) readFromRemoteNode(address, key string) (api.GetResponse, error) { + url := fmt.Sprintf("http://%s/internal/storage/%s", address, key) + resp, err := s.client.Get(url) + if err != nil { + return api.GetResponse{}, err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return api.GetResponse{}, fmt.Errorf("remote node returned status %d", resp.StatusCode) + } + + var result api.ReplicateGetResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return api.GetResponse{}, err + } + return api.GetResponse{ + Key: result.Key, + Value: result.Value, + Found: result.Found, + }, nil +} diff --git a/pkg/api/types.go b/pkg/api/types.go index 7011c65..ba98b61 100644 --- a/pkg/api/types.go +++ b/pkg/api/types.go @@ -17,3 +17,27 @@ type GetResponse struct { Versions []map[string]uint64 `json:"versions,omitempty"` Found bool `json:"found"` } + +// Internal replication types + +type ReplicateRequest struct { + Key string `json:"key"` + Value []byte `json:"value"` + Version map[string]uint64 `json:"version"` +} + +type ReplicateResponse struct { + Success bool `json:"success"` + Error string `json:"error,omitempty"` +} + +type ReplicateGetRequest struct { + Key string `json:"key"` +} + +type ReplicateGetResponse struct { + Key string `json:"key"` + Value []byte `json:"value,omitempty"` + Version map[string]uint64 `json:"version,omitempty"` + Found bool `json:"found"` +}