Support queuing score by id

README.md

@@ -16,8 +16,8 @@ A string value is used to indicate the current schema version to be used.
 
 ## Adding items to be indexed
 
-Scores with `preserve`=`true` will be added to the index,
-scores with `preserve`=`false` will be removed from the index.
+Scores with `preserve`=`true` belonging to a user with `user_warnings`=`0` will be added to the index,
+scores where any of the previous conditions are false will be removed from the index.
 
 Push items to `osu-queue:score-index-${schema}`
 
@@ -124,6 +124,14 @@ For testing purposes, we can add fake items to the queue:
 
 It should be noted that these items will not exist or match the ones in the database.
 
+## Queuing a specific score for indexing
+
+    schema=${schema} dotnet run index ${id}
+
+will queue the score with `${id}` for indexing; the score will be added or deleted as necessary, according to the value of `SoloScore.ShouldIndex`.
+
+See [Queuing items for processing from another client](#queuing-items-for-processing-from-another-client)
+
 ## Adding existing database records to the queue
 
     schema=1 dotnet run all
@@ -149,3 +157,30 @@ Populating an index is done by pushing score items to a queue.
 where `${cmd}` is the command to run, e.g. `dotnet osu.ElasticIndexer.dll queue`
 
 # Typical use-cases
+
+## Queuing items for processing from another client
+
+Push items into the Redis queue "`osu-queue:score-index-${schema}`"
+e.g.
+
+```csharp
+ListLeftPush("osu-queue:score-index-1", "{ \"ScoreId\": 1 }");
+```
+
+or from redis-cli:
+```
+LPUSH "osu-queue:score-index-1" "{\"ScoreId\":1}"
+```
+
+### Indexing a score by `id`
+```json
+{ "ScoreId": 1 }
+```
+
+### Queuing a whole score
+
+```json
+{
+    "Score": {Solo.Score}
+}
+```

osu.ElasticIndexer/Commands/IndexCommand.cs

@@ -0,0 +1,32 @@
+// Copyright (c) ppy Pty Ltd <[email protected]>. Licensed under the MIT Licence.
+// See the LICENCE file in the repository root for full licence text.
+
+using System;
+using System.ComponentModel.DataAnnotations;
+using System.Threading;
+using McMaster.Extensions.CommandLineUtils;
+
+namespace osu.ElasticIndexer.Commands
+{
+    [Command("index", Description = "Queue a score for indexing by id.")]
+    public class IndexCommand : ProcessorCommandBase
+    {
+        [Argument(1)]
+        [Required]
+        public string ScoreId { get; set; } = string.Empty;
+
+        public int OnExecute(CancellationToken cancellationToken)
+        {
+            if (string.IsNullOrEmpty(AppSettings.Schema))
+                throw new MissingSchemaException();
+
+            var id = long.Parse(ScoreId);
+            var scoreItem = new ScoreItem { ScoreId = id };
+            Processor.PushToQueue(scoreItem);
+
+            Console.WriteLine(ConsoleColor.Cyan, $"Queued to {Processor.QueueName}: {scoreItem}");
+
+            return 0;
+        }
+    }
+}

osu.ElasticIndexer/Commands/ListIndicesCommand.cs

@@ -26,7 +26,7 @@ public int OnExecute()
                     var schema = indexState.Mappings.Meta?["schema"];
                     var aliased = indexState.Aliases.ContainsKey(client.AliasName);
 
-                    Console.WriteLine($"{record.Index} schema:{schema} aliased:{aliased} {record.Status} {record.DocsCount} {record.PrimaryStoreSize}");
+                    Console.WriteLine($"{record.Index} schema:{schema} aliased:{aliased} {record.Status} docs {record.DocsCount} deleted {record.DocsDeleted} {record.PrimaryStoreSize}");
                 }
             }
 

osu.ElasticIndexer/Commands/PumpAllScoresCommand.cs

@@ -53,7 +53,7 @@ public int OnExecute(CancellationToken cancellationToken)
                     if (Verbose)
                         Console.WriteLine($"Pushing {score}");
 
-                    Processor.PushToQueue(new ScoreItem(score));
+                    Processor.PushToQueue(new ScoreItem { Score = score });
                 }
 
                 if (!Verbose)

osu.ElasticIndexer/Commands/PumpFakeScoresCommand.cs

@@ -54,7 +54,7 @@ public int OnExecute(CancellationToken cancellationToken)
                         preserve = true
                     };
 
-                Processor.PushToQueue(new ScoreItem(score));
+                Processor.PushToQueue(new ScoreItem { Score = score });
 
                 if (counter % 1000 == 0)
                     Console.WriteLine($"pushed to {Processor.QueueName}, current id: {counter}");

osu.ElasticIndexer/Commands/PushFileCommand.cs

@@ -0,0 +1,31 @@
+// Copyright (c) ppy Pty Ltd <[email protected]>. Licensed under the MIT Licence.
+// See the LICENCE file in the repository root for full licence text.
+
+using System.ComponentModel.DataAnnotations;
+using System.IO;
+using System.Threading;
+using McMaster.Extensions.CommandLineUtils;
+
+namespace osu.ElasticIndexer.Commands
+{
+    [Command("push-file", Description = "Push contents of a file to the queue for testing.")]
+    public class PushFileCommand : ProcessorCommandBase
+    {
+        [Argument(0)]
+        [Required]
+        public string Filename { get; set; } = string.Empty;
+
+        public int OnExecute(CancellationToken cancellationToken)
+        {
+            if (string.IsNullOrEmpty(AppSettings.Schema))
+                throw new MissingSchemaException();
+
+            var value = File.ReadAllText(Filename);
+            var redis = new Redis();
+
+            redis.Connection.GetDatabase().ListLeftPush(Processor.QueueName, value);
+
+            return 0;
+        }
+    }
+}

osu.ElasticIndexer/ElasticModel.cs

@@ -61,5 +61,21 @@ public static IEnumerable<List<T>> Chunk<T>(string? where, int chunkSize = 10000
 
         public static IEnumerable<List<T>> Chunk<T>(int chunkSize = 10000, long? resumeFrom = null) where T : ElasticModel =>
             Chunk<T>(null, chunkSize, resumeFrom);
+
+        public static List<T> Find<T>(IEnumerable<long> ids) where T : ElasticModel
+        {
+            using (var dbConnection = new MySqlConnection(AppSettings.ConnectionString))
+            {
+                var attribute = typeof(T).GetCustomAttributes<ChunkOnAttribute>().First();
+                var cursorColumn = attribute.CursorColumn;
+                var selects = attribute.Query;
+
+                string query = $"select {selects} where {cursorColumn} in @ids";
+
+                dbConnection.Open();
+
+                return dbConnection.Query<T>(query, new { ids }).AsList();
+            }
+        }
     }
 }

osu.ElasticIndexer/IndexQueueProcessor.cs

@@ -3,9 +3,9 @@
 
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
 using System.Threading.Tasks;
-using Elasticsearch.Net;
 using Nest;
 using osu.Server.QueueProcessor;
 
@@ -38,41 +38,44 @@ internal IndexQueueProcessor(string index, Client client, Action stopCallback)
 
         protected override void ProcessResults(IEnumerable<ScoreItem> items)
         {
-            var add = new List<SoloScore>();
-            var remove = new List<SoloScore>();
+            var buffer = new ProcessableItemsBuffer();
 
+            // Figure out what to do with the queue item.
             foreach (var item in items)
             {
-                if (item.Score.ShouldIndex)
-                    add.Add(item.Score);
+                if (item.ScoreId != null)
+                {
+                    buffer.ScoreIdsForLookup.Add(item.ScoreId.Value);
+                }
+                else if (item.Score != null)
+                {
+                    if (item.Score.ShouldIndex)
+                        buffer.Additions.Add(item.Score);
+                    else
+                        buffer.Deletions.Add(item.Score.id);
+                }
                 else
-                    remove.Add(item.Score);
+                {
+                    Console.WriteLine(ConsoleColor.Red, "queue item missing both data and action");
+                }
             }
 
-            var bulkDescriptor = new BulkDescriptor()
-                                 .Index(index)
-                                 .IndexMany(add)
-                                 .DeleteMany(remove);
+            // Handle any scores that need a lookup from the database.
+            performDatabaseLookup(buffer);
 
-            var response = client.ElasticClient.Bulk(bulkDescriptor);
+            Debug.Assert(buffer.ScoreIdsForLookup.Count == 0);
 
-            handleResponse(response, items);
-        }
-
-        private BulkResponse dispatch(BulkDescriptor bulkDescriptor)
-        {
-            try
-            {
-                return client.ElasticClient.Bulk(bulkDescriptor);
-            }
-            catch (ElasticsearchClientException ex)
+            if (buffer.Additions.Any() || buffer.Deletions.Any())
             {
-                // Server disappeared, maybe network failure or it's restarting; spin until it's available again.
-                Console.WriteLine(ConsoleColor.Red, ex.Message);
-                Console.WriteLine(ConsoleColor.Yellow, ex.InnerException?.Message);
-                waitUntilActive();
+                var bulkDescriptor = new BulkDescriptor()
+                                     .Index(index)
+                                     .IndexMany(buffer.Additions)
+                                     // type is needed for ids https://github.com/elastic/elasticsearch-net/issues/3500
+                                     .DeleteMany<SoloScore>(buffer.Deletions);
 
-                return dispatch(bulkDescriptor);
+                var response = client.ElasticClient.Bulk(bulkDescriptor);
+
+                handleResponse(response, items);
             }
         }
 
@@ -95,7 +98,7 @@ private void handleResponse(BulkResponse response, IEnumerable<ScoreItem> items)
             // Elasticsearch bulk thread pool is full.
             if (response.ItemsWithErrors.Any(item => item.Status == 429 || item.Error.Type == "es_rejected_execution_exception"))
             {
-                Console.WriteLine(ConsoleColor.Yellow, $"Server returned 429, re-queued chunk with lastId {items.Last().Score.id}");
+                Console.WriteLine(ConsoleColor.Yellow, $"Server returned 429, re-queued chunk with lastId {items.Last()}");
 
                 foreach (var item in items)
                 {
@@ -124,6 +127,27 @@ private void handleResponse(BulkResponse response, IEnumerable<ScoreItem> items)
             // TODO: per-item errors?
         }
 
+        private void performDatabaseLookup(ProcessableItemsBuffer buffer)
+        {
+            if (!buffer.ScoreIdsForLookup.Any()) return;
+
+            var scores = ElasticModel.Find<SoloScore>(buffer.ScoreIdsForLookup);
+
+            foreach (var score in scores)
+            {
+                if (score.ShouldIndex)
+                    buffer.Additions.Add(score);
+                else
+                    buffer.Deletions.Add(score.id);
+
+                buffer.ScoreIdsForLookup.Remove(score.id);
+            }
+
+            // Remaining scores do not exist and should be deleted.
+            buffer.Deletions.AddRange(buffer.ScoreIdsForLookup);
+            buffer.ScoreIdsForLookup.Clear();
+        }
+
         private void waitUntilActive()
         {
             // Spin until valid response from elasticsearch.

osu.ElasticIndexer/ProcessableItemsBuffer.cs

@@ -0,0 +1,26 @@
+// Copyright (c) ppy Pty Ltd <[email protected]>. Licensed under the MIT Licence.
+// See the LICENCE file in the repository root for full licence text.
+
+using System.Collections.Generic;
+
+namespace osu.ElasticIndexer
+{
+    public class ProcessableItemsBuffer
+    {
+        /// <summary>
+        /// A set of all score IDs which have arrived but are not yet determined to be an addition or deletion.
+        /// These should be processed into either <see cref="Additions"/> or <see cref="Deletions"/>.
+        /// </summary>
+        public readonly HashSet<long> ScoreIdsForLookup = new HashSet<long>();
+
+        /// <summary>
+        /// New scores which should be indexed.
+        /// </summary>
+        public readonly List<SoloScore> Additions = new List<SoloScore>();
+
+        /// <summary>
+        /// Score IDs which should be purged from the index is they are present.
+        /// </summary>
+        public readonly List<long> Deletions = new List<long>();
+    }
+}

osu.ElasticIndexer/Program.cs

@@ -12,9 +12,11 @@ namespace osu.ElasticIndexer
     [Subcommand(typeof(CloseIndexCommand))]
     [Subcommand(typeof(DeleteIndexCommand))]
     [Subcommand(typeof(ListIndicesCommand))]
+    [Subcommand(typeof(IndexCommand))]
     [Subcommand(typeof(OpenIndexCommand))]
     [Subcommand(typeof(PumpAllScoresCommand))]
     [Subcommand(typeof(PumpFakeScoresCommand))]
+    [Subcommand(typeof(PushFileCommand))]
     [Subcommand(typeof(SchemaVersionCommand))]
     [Subcommand(typeof(UpdateAliasCommand))]
     [Subcommand(typeof(WatchQueueCommand))]

osu.ElasticIndexer/ScoreItem.cs

@@ -7,13 +7,12 @@ namespace osu.ElasticIndexer
 {
     public class ScoreItem : QueueItem
     {
-        public SoloScore Score { get; }
+        public long? ScoreId { get; init; }
 
-        public ScoreItem(SoloScore score)
-        {
-            Score = score;
-        }
+        // ScoreId is always preferred if present (this property is ignored).
+        // Note that this is generally not used anymore. Consider removing in the future unless a use case comes up?
+        public SoloScore? Score { get; init; }
 
-        public override string ToString() => $"ScoreItem id: {Score.id}";
+        public override string ToString() => Score != null ? $"ScoreItem Score: {Score.id}" : $"ScoreItem ScoreId: {ScoreId}";
     }
 }

osu.ElasticIndexer/SoloScore.cs

@@ -69,6 +69,10 @@ public string data
         [Number(NumberType.Integer)]
         public int? build_id => scoreData.build_id;
 
+        [Computed]
+        [Boolean]
+        public bool convert { get; set; }
+
         [Computed]
         [Boolean]
         public bool passed => scoreData.passed;

osu.ElasticIndexer/schemas/solo_scores.json

@@ -14,6 +14,9 @@
       "build_id": {
         "type": "integer"
       },
+      "convert": {
+        "type": "boolean"
+      },
       "country_code": {
         "type": "keyword"
       },