defmodule IsLabDB.GravitationalRouter do @moduledoc """ Physics-Based Data Routing Engine for Spacetime Shards This module implements intelligent data routing using gravitational attraction calculations, consistent hashing, and access pattern analysis. It determines the optimal spacetime shard for each piece of data based on: - Gravitational attraction between data and shards - Current load distribution and capacity - Access pattern compatibility - Locality-sensitive routing for related data - Conservation of energy during data placement ## Physics Concepts - **Gravitational Routing**: Data naturally "falls" into shards with stronger attraction - **Consistent Hashing**: Ensures even distribution while allowing for shard expansion - **Locality Sensitivity**: Related data items are placed in nearby cosmic regions - **Load Balancing**: Prevents gravitational collapse by distributing mass evenly - **Conservation Laws**: Energy and data are conserved during routing operations """ require Logger alias IsLabDB.{SpacetimeShard} defstruct [ :shard_topology, # Map of available shards :routing_algorithm, # Current routing algorithm (:gravitational, :consistent_hash, etc.) :locality_map, # Spatial relationships between data :load_balancer, # Load balancing configuration :migration_manager, # Data migration coordination :routing_cache, # Cached routing decisions :performance_metrics, # Routing performance tracking :consistency_coordinator # Cross-shard consistency management ] ## PUBLIC API @doc """ Initialize the gravitational routing system. ## Parameters - `shards` - List of spacetime shards to manage - `opts` - Router configuration options ## Options - `:routing_algorithm` - Primary routing algorithm (default: :gravitational) - `:cache_size` - Routing decision cache size (default: 1000) - `:rebalancing_threshold` - Load imbalance threshold (default: 0.3) - `:locality_range` - Locality sensitivity range (default: 100.0) ## Examples {:ok, router} = GravitationalRouter.initialize([hot_shard, warm_shard, cold_shard]) """ def initialize(shards, opts \\ []) do Logger.info("🌌 Initializing gravitational routing system with #{length(shards)} shards") # Create shard topology map shard_topology = Enum.reduce(shards, %{}, fn shard, acc -> Map.put(acc, shard.shard_id, shard) end) # Initialize routing components router = %__MODULE__{ shard_topology: shard_topology, routing_algorithm: Keyword.get(opts, :routing_algorithm, :gravitational), locality_map: initialize_locality_map(), load_balancer: initialize_load_balancer(opts), migration_manager: initialize_migration_manager(), routing_cache: initialize_routing_cache(opts), performance_metrics: initialize_performance_metrics(), consistency_coordinator: initialize_consistency_coordinator() } # Start background processes start_background_processes(router) Logger.info("✨ Gravitational router initialized with #{router.routing_algorithm} algorithm") {:ok, router} end @doc """ Determine the optimal shard for storing data using gravitational calculations. This is the main routing function that analyzes all available shards and selects the one with the strongest gravitational attraction for the data. ## Parameters - `router` - The gravitational router instance - `key` - Data key to route - `value` - Data value - `opts` - Routing options and metadata ## Options - `:access_pattern` - Expected access pattern (:hot, :warm, :cold) - `:priority` - Data priority level (:critical, :high, :normal, :low) - `:locality_group` - Locality group for related data placement - `:force_shard` - Force placement in specific shard (bypasses routing) ## Returns `{:ok, shard_id, routing_metadata}` or `{:error, reason}` ## Examples {:ok, :hot_data, metadata} = GravitationalRouter.route_data(router, "user:alice", user_data, access_pattern: :hot, priority: :critical) """ def route_data(router, key, value, opts \\ []) do start_time = :os.system_time(:microsecond) # Check for forced shard placement case Keyword.get(opts, :force_shard) do nil -> # Perform gravitational routing perform_gravitational_routing(router, key, value, opts, start_time) forced_shard when is_atom(forced_shard) -> # Validate forced shard exists and accept it case Map.get(router.shard_topology, forced_shard) do nil -> {:error, {:invalid_shard, forced_shard}} _shard -> end_time = :os.system_time(:microsecond) metadata = %{ routing_algorithm: :forced, operation_time: end_time - start_time, gravitational_score: 0.0, shard_id: forced_shard } {:ok, forced_shard, metadata} end end end @doc """ Analyze current load distribution across all shards. Returns comprehensive analysis of load balancing, gravitational field distribution, and recommendations for optimization. ## Returns Map containing load analysis and rebalancing recommendations """ def analyze_load_distribution(router) do shard_loads = Enum.map(router.shard_topology, fn {shard_id, shard} -> metrics = SpacetimeShard.get_shard_metrics(shard) {shard_id, metrics} end) total_items = Enum.reduce(shard_loads, 0, fn {_id, metrics}, acc -> acc + metrics.data_items end) # Calculate load imbalance load_imbalance = calculate_load_imbalance(shard_loads, total_items) # Detect gravitational hot spots gravitational_hotspots = detect_gravitational_hotspots(shard_loads) # Generate rebalancing recommendations recommendations = generate_rebalancing_recommendations(shard_loads, load_imbalance) %{ total_data_items: total_items, shard_distribution: shard_loads, load_imbalance_factor: load_imbalance, gravitational_hotspots: gravitational_hotspots, rebalancing_needed: load_imbalance > router.load_balancer.imbalance_threshold, recommendations: recommendations, analysis_timestamp: :os.system_time(:millisecond) } end @doc """ Execute load rebalancing across shards using gravitational migration. Automatically migrates data between shards to achieve better load distribution while respecting gravitational attraction principles. ## Parameters - `router` - The gravitational router - `analysis` - Load analysis from `analyze_load_distribution/1` - `opts` - Rebalancing options ## Options - `:migration_batch_size` - Number of items to migrate per batch (default: 100) - `:max_migration_time` - Maximum time for rebalancing operation (default: 30_000ms) - `:dry_run` - Only plan migrations without executing (default: false) ## Returns `{:ok, rebalancing_results}` or `{:error, reason}` """ def execute_gravitational_rebalancing(router, analysis, opts \\ []) do Logger.info("🌌 Starting gravitational rebalancing operation") dry_run = Keyword.get(opts, :dry_run, false) batch_size = Keyword.get(opts, :migration_batch_size, 100) max_time = Keyword.get(opts, :max_migration_time, 30_000) start_time = :os.system_time(:millisecond) # Create migration plan based on analysis migration_plan = create_migration_plan(analysis, batch_size) if dry_run do Logger.info("📋 Dry run: Would migrate #{length(migration_plan.migrations)} data groups") {:ok, %{plan: migration_plan, executed: false}} else # Execute migration plan with time limit execute_migration_plan(router, migration_plan, start_time, max_time) end end @doc """ Find all data items that should be co-located based on locality groups. Analyzes data relationships and identifies items that would benefit from gravitational clustering in the same or nearby shards. ## Parameters - `router` - The gravitational router - `locality_group` - Locality group identifier - `opts` - Search options ## Returns List of related data items and their optimal placement recommendations """ def find_locality_clusters(router, locality_group, opts \\ []) do max_distance = Keyword.get(opts, :max_distance, router.load_balancer.locality_range) # Search for data in the specified locality group locality_items = search_locality_group(router, locality_group) # Calculate optimal clustering clusters = calculate_optimal_clustering(locality_items, max_distance) # Generate placement recommendations Enum.map(clusters, fn cluster -> optimal_shard = calculate_cluster_optimal_shard(router, cluster) %{ cluster_id: cluster.id, items: cluster.items, current_distribution: cluster.current_distribution, recommended_shard: optimal_shard, clustering_score: cluster.cohesion_score } end) end @doc """ Get comprehensive routing performance metrics. ## Returns Map containing routing performance statistics and efficiency metrics """ def get_routing_metrics(router) do current_time = :os.system_time(:millisecond) base_metrics = %{ total_routing_decisions: router.performance_metrics.total_decisions, cache_hit_rate: calculate_cache_hit_rate(router), average_routing_time: router.performance_metrics.avg_decision_time, algorithm_efficiency: calculate_algorithm_efficiency(router), load_balance_score: calculate_load_balance_score(router), gravitational_field_strength: calculate_total_gravitational_strength(router) } # Add per-shard routing statistics shard_stats = Enum.map(router.shard_topology, fn {shard_id, shard} -> try do shard_metrics = SpacetimeShard.get_shard_metrics(shard) routing_stats = get_shard_routing_stats(router, shard_id) {shard_id, Map.merge(shard_metrics, routing_stats)} rescue error -> Logger.warning("Failed to get metrics for shard #{shard_id}: #{inspect(error)}") # Return basic fallback metrics {shard_id, %{ shard_id: shard_id, data_items: 0, memory_usage: 0, gravitational_field_strength: 0.0, entropy_level: 0.0 }} end end) |> Enum.into(%{}) Map.merge(base_metrics, %{ shard_statistics: shard_stats, last_updated: current_time }) end ## PRIVATE FUNCTIONS defp perform_gravitational_routing(router, key, value, opts, start_time) do # Check routing cache first cache_key = generate_cache_key(key, value, opts) # Check routing cache - currently always returns :miss (simplified implementation) case get_cached_routing_decision(router, cache_key) do :miss -> # Perform full gravitational calculation perform_full_gravitational_analysis(router, key, value, opts, start_time) # Note: In full implementation, would handle {:hit, cached_decision} case here end end defp perform_full_gravitational_analysis(router, key, value, opts, start_time) do access_metadata = %{ access_pattern: Keyword.get(opts, :access_pattern, :balanced), priority: Keyword.get(opts, :priority, :normal), locality_group: Keyword.get(opts, :locality_group), access_frequency: Keyword.get(opts, :access_frequency, 1.0) } # Check for empty shard topology if map_size(router.shard_topology) == 0 do {:error, :no_shards_available} else # Calculate gravitational scores for all shards shard_scores = Enum.map(router.shard_topology, fn {shard_id, shard} -> score = SpacetimeShard.calculate_gravitational_score(shard, key, value, access_metadata) {shard_id, score, shard} end) # Apply consistent hashing for tie-breaking and stability shard_scores_with_hash = Enum.map(shard_scores, fn {shard_id, score, shard} -> hash_score = :erlang.phash2({key, shard_id}, 1000) / 1000.0 adjusted_score = score * (1.0 + hash_score * 0.1) # Small hash-based adjustment {shard_id, adjusted_score, shard} end) # Find the shard with highest gravitational attraction {best_shard_id, best_score, _best_shard} = Enum.max_by(shard_scores_with_hash, fn {_id, score, _shard} -> score end) end_time = :os.system_time(:microsecond) operation_time = end_time - start_time routing_metadata = %{ routing_algorithm: router.routing_algorithm, operation_time: operation_time, gravitational_score: best_score, shard_id: best_shard_id, all_scores: Enum.map(shard_scores_with_hash, fn {id, score, _} -> %{shard_id: id, score: Float.round(score, 3)} end), cache_hit: false } # Cache this routing decision cache_routing_decision(router, generate_cache_key(key, value, opts), best_shard_id, routing_metadata) # Update routing performance metrics update_routing_metrics(router, :routing_decision, operation_time) Logger.debug("🎯 Routed #{key} to #{best_shard_id} (score: #{Float.round(best_score, 3)})") {:ok, best_shard_id, routing_metadata} end end defp initialize_locality_map() do %{ groups: %{}, spatial_index: %{}, relationship_graph: %{} } end defp initialize_load_balancer(opts) do %{ imbalance_threshold: Keyword.get(opts, :rebalancing_threshold, 0.3), locality_range: Keyword.get(opts, :locality_range, 100.0), migration_batch_size: Keyword.get(opts, :migration_batch_size, 100), rebalancing_interval: Keyword.get(opts, :rebalancing_interval, 300_000), last_rebalancing: :os.system_time(:millisecond) } end defp initialize_migration_manager() do %{ active_migrations: %{}, migration_history: [], migration_locks: %{} } end defp initialize_routing_cache(opts) do cache_size = Keyword.get(opts, :cache_size, 1000) %{ decisions: %{}, access_order: [], max_size: cache_size, hit_count: 0, miss_count: 0 } end defp initialize_performance_metrics() do %{ total_decisions: 0, total_decision_time: 0, avg_decision_time: 0.0, algorithm_switches: 0, cache_hits: 0, cache_misses: 0 } end defp initialize_consistency_coordinator() do %{ cross_shard_operations: %{}, consistency_locks: %{}, transaction_log: [] } end defp start_background_processes(_router) do # Start background processes for maintenance # This would include periodic rebalancing, cache cleanup, etc. Logger.debug("🔄 Background routing processes initialized") :ok end defp calculate_load_imbalance(shard_loads, total_items) when total_items > 0 do # Calculate coefficient of variation for load distribution shard_count = length(shard_loads) expected_load = total_items / shard_count variance = Enum.reduce(shard_loads, 0.0, fn {_id, metrics}, acc -> deviation = metrics.data_items - expected_load acc + (deviation * deviation) end) / shard_count standard_deviation = :math.sqrt(variance) # Return coefficient of variation (0 = perfect balance, 1+ = very imbalanced) if expected_load > 0, do: standard_deviation / expected_load, else: 0.0 end defp calculate_load_imbalance(_shard_loads, 0), do: 0.0 defp detect_gravitational_hotspots(shard_loads) do # Find shards with unusually high gravitational field strength total_gravitational_strength = Enum.reduce(shard_loads, 0.0, fn {_id, metrics}, acc -> acc + metrics.gravitational_field_strength end) average_strength = total_gravitational_strength / length(shard_loads) Enum.filter(shard_loads, fn {_shard_id, metrics} -> metrics.gravitational_field_strength > average_strength * 1.5 end) |> Enum.map(fn {shard_id, metrics} -> %{ shard_id: shard_id, field_strength: metrics.gravitational_field_strength, severity: metrics.gravitational_field_strength / average_strength } end) end defp generate_rebalancing_recommendations(shard_loads, load_imbalance) do if load_imbalance > 0.3 do # Find overloaded and underloaded shards total_items = Enum.reduce(shard_loads, 0, fn {_id, metrics}, acc -> acc + metrics.data_items end) average_load = total_items / length(shard_loads) overloaded = Enum.filter(shard_loads, fn {_id, metrics} -> metrics.data_items > average_load * 1.2 end) underloaded = Enum.filter(shard_loads, fn {_id, metrics} -> metrics.data_items < average_load * 0.8 end) migration_recommendations = Enum.map(overloaded, fn {source_shard, metrics} -> excess_items = round(metrics.data_items - average_load) target_shard = find_best_migration_target(underloaded, excess_items) %{ action: :migrate, from: source_shard, to: target_shard, estimated_items: excess_items, urgency: :medium } end) migration_recommendations else [] end end defp find_best_migration_target(underloaded_shards, _excess_items) do # Find the shard with the most available capacity case underloaded_shards do [] -> nil [{shard_id, _metrics}] -> shard_id shards -> {best_shard_id, _} = Enum.min_by(shards, fn {_id, metrics} -> metrics.data_items end) best_shard_id end end defp create_migration_plan(analysis, batch_size) do migrations = Enum.map(analysis.recommendations, fn recommendation -> %{ source_shard: recommendation.from, target_shard: recommendation.to, batch_size: batch_size, estimated_items: recommendation.estimated_items, urgency: recommendation.urgency } end) %{ migrations: migrations, estimated_duration: length(migrations) * 5_000, # 5 seconds per migration batch total_items: Enum.reduce(migrations, 0, fn m, acc -> acc + m.estimated_items end) } end defp execute_migration_plan(router, migration_plan, start_time, max_time) do Logger.info("🚀 Executing migration plan: #{length(migration_plan.migrations)} migrations") results = Enum.map(migration_plan.migrations, fn migration -> current_time = :os.system_time(:millisecond) elapsed_time = current_time - start_time if elapsed_time > max_time do Logger.warning("⏰ Migration timeout reached, stopping") {:error, :timeout} else execute_single_migration(router, migration) end end) successful_migrations = Enum.count(results, fn result -> case result do {:ok, _} -> true _ -> false end end) end_time = :os.system_time(:millisecond) total_time = end_time - start_time Logger.info("✅ Migration completed: #{successful_migrations}/#{length(migration_plan.migrations)} successful") {:ok, %{ successful_migrations: successful_migrations, failed_migrations: length(results) - successful_migrations, total_duration_ms: total_time, migration_results: results }} end defp execute_single_migration(router, migration) do source_shard = Map.get(router.shard_topology, migration.source_shard) target_shard = Map.get(router.shard_topology, migration.target_shard) if source_shard && target_shard do SpacetimeShard.initiate_data_migration(source_shard, target_shard, %{ batch_size: migration.batch_size, max_items: migration.estimated_items }) else {:error, :invalid_shards} end end defp search_locality_group(_router, _locality_group) do # Placeholder for locality group search [] end defp calculate_optimal_clustering(_locality_items, _max_distance) do # Placeholder for clustering calculation [] end defp calculate_cluster_optimal_shard(_router, _cluster) do # Placeholder for cluster optimal shard calculation :hot_data end defp generate_cache_key(key, value, opts) do :crypto.hash(:md5, :erlang.term_to_binary({key, :erlang.phash2(value), opts})) |> Base.encode16(case: :lower) end defp get_cached_routing_decision(_router, _cache_key) do # Simplified cache implementation - always returns miss for now # In future versions, this could implement actual caching logic :miss end defp cache_routing_decision(_router, _cache_key, _shard_id, _metadata) do # Simplified cache implementation :ok end defp update_routing_metrics(_router, _operation_type, _operation_time) do # Update routing performance metrics :ok end defp calculate_cache_hit_rate(router) do total_requests = router.performance_metrics.cache_hits + router.performance_metrics.cache_misses if total_requests > 0 do router.performance_metrics.cache_hits / total_requests else 0.0 end end defp calculate_algorithm_efficiency(_router) do # Placeholder for algorithm efficiency calculation 0.85 end defp calculate_load_balance_score(_router) do # Placeholder for load balance score calculation 0.92 end defp calculate_total_gravitational_strength(router) do Enum.reduce(router.shard_topology, 0.0, fn {_id, shard}, acc -> try do metrics = SpacetimeShard.get_shard_metrics(shard) field_strength = metrics.gravitational_field_strength || 0.0 acc + field_strength rescue _ -> acc # Skip shards that can't provide metrics end end) end defp get_shard_routing_stats(_router, _shard_id) do # Placeholder for shard-specific routing statistics %{ routing_decisions: 0, avg_routing_score: 0.0, migration_events: 0 } end end