Fix races in rename

src/main/scala/com/cloudant/clouseau/ClouseauTypeFactory.scala

@@ -30,7 +30,7 @@ case class SearchRequest(options: Map[Symbol, Any])
 
 case class OpenIndexMsg(peer: Pid, path: String, options: Any)
 case class CleanupPathMsg(path: String)
-case class RenamePathMsg(dbName: String)
+case class RenamePathMsg(path: String)
 case class CleanupDbMsg(dbName: String, activeSigs: List[String])
 case class DiskSizeMsg(path: String)
 case class CloseLRUByPathMsg(path: String)

src/main/scala/com/cloudant/clouseau/IndexCleanupService.scala

@@ -31,19 +31,27 @@ class IndexCleanupService(ctx: ServiceContext[ConfigurationArgs]) extends Servic
       val dir = new File(rootDir, path)
       logger.info("Removing %s".format(path))
       recursivelyDelete(dir)
-    case RenamePathMsg(dbName: String) =>
-      val srcDir = new File(rootDir, dbName)
+    case RenamePathMsg(path: String) =>
+      val srcDir = new File(rootDir, path)
       val sdf = new SimpleDateFormat("yyyyMMdd'.'HHmmss")
       sdf.setTimeZone(TimeZone.getTimeZone("UTC"))
       val sdfNow = sdf.format(Calendar.getInstance().getTime())
-      // move timestamp information in dbName to end of destination path
+      // move timestamp information in path to end of destination path
       // for example, from foo.1234567890 to foo.20170912.092828.deleted.1234567890
-      val destPath = dbName.dropRight(10) + sdfNow + ".deleted." + dbName.takeRight(10)
+      val destPath = path.dropRight(10) + sdfNow + ".deleted." + path.takeRight(10)
       val destDir = new File(rootDir, destPath)
-      logger.info("Renaming '%s' to '%s'".format(
-        srcDir.getAbsolutePath, destDir.getAbsolutePath)
-      )
-      rename(srcDir, destDir)
+      destDir.mkdirs()
+
+      for (index <- srcDir.listFiles) {
+        val indexPath = path + File.separator + index.getName
+        val renameDir = new File(destDir, index.getName)
+        call('main, ('rename, indexPath, renameDir.getAbsolutePath)) match {
+          case 'ok =>
+            'ok
+          case ('error, 'not_found) =>
+            rename(new File(srcDir, index.getName), renameDir)
+        }
+      }
     case CleanupDbMsg(dbName: String, activeSigs: List[String]) =>
       logger.info("Cleaning up " + dbName)
       val pattern = Pattern.compile("shards/[0-9a-f]+-[0-9a-f]+/" + dbName + "\\.[0-9]+/([0-9a-f]+)$")
@@ -82,6 +90,9 @@ class IndexCleanupService(ctx: ServiceContext[ConfigurationArgs]) extends Servic
     if (!srcDir.isDirectory) {
       return
     }
+    logger.info("Renaming closed index '%s' to '%s'".format(
+        srcDir.getAbsolutePath, destDir.getAbsolutePath)
+      )
     if (!srcDir.renameTo(destDir)) {
       logger.error("Failed to rename directory from '%s' to '%s'".format(
         srcDir.getAbsolutePath, destDir.getAbsolutePath))

src/main/scala/com/cloudant/clouseau/IndexManagerService.scala

@@ -138,6 +138,14 @@ class IndexManagerService(ctx: ServiceContext[ConfigurationArgs]) extends Servic
       'ok
     case 'version =>
       ('ok, getClass.getPackage.getImplementationVersion)
+    case ('rename, indexPath: String, destPath: String) =>
+      lru.get(indexPath) match {
+        case null =>
+          ('error, 'not_found)
+        case pid: Pid =>
+          pid ! ('rename, destPath)
+          'ok
+      }
   }
 
   override def handleInfo(msg: Any) = msg match {

src/main/scala/com/cloudant/clouseau/IndexService.scala

@@ -58,7 +58,7 @@ import com.spatial4j.core.context.SpatialContext
 import com.spatial4j.core.distance.DistanceUtils
 import java.util.HashSet
 
-case class IndexServiceArgs(config: Configuration, name: String, queryParser: QueryParser, writer: IndexWriter)
+case class IndexServiceArgs(config: Configuration, rootDir: File, name: String, queryParser: QueryParser, writer: IndexWriter)
 case class HighlightParameters(highlighter: Highlighter, highlightFields: List[String], highlightNumber: Int, analyzers: List[Analyzer])
 
 // These must match the records in dreyfus.
@@ -169,6 +169,18 @@ class IndexService(ctx: ServiceContext[IndexServiceArgs]) extends Service(ctx) w
         exit("Idle Timeout")
       }
       idle = true
+    case ('rename, path: String) =>
+      close()
+      val srcDir = new File(ctx.args.rootDir, ctx.args.name)
+      val destDir = new File(path)
+      info("Renaming open index '%s' to '%s'".format(
+        srcDir.getAbsolutePath, destDir.getAbsolutePath)
+      )
+      if (!srcDir.renameTo(destDir)) {
+        error("Failed to rename directory from '%s' to '%s'".format(
+          srcDir.getAbsolutePath, destDir.getAbsolutePath))
+      }
+      exit('close)
     case 'count_fields =>
       countFields
     case 'delete =>
@@ -216,15 +228,24 @@ class IndexService(ctx: ServiceContext[IndexServiceArgs]) extends Service(ctx) w
     } catch {
       case e: IOException => warn("Error while closing reader", e)
     }
+    try {
+      close()
+    } finally {
+      super.exit(msg)
+    }
+  }
+
+  private def close() {
     try {
       ctx.args.writer.rollback()
     } catch {
       case e: AlreadyClosedException => 'ignored
       case e: IOException =>
         warn("Error while closing writer", e)
-        ctx.args.writer.close()
-    } finally {
-      super.exit(msg)
+        val dir = ctx.args.writer.getDirectory
+        if (IndexWriter.isLocked(dir)) {
+          IndexWriter.unlock(dir);
+        }
     }
   }
 
@@ -853,6 +874,10 @@ class IndexService(ctx: ServiceContext[IndexServiceArgs]) extends Service(ctx) w
     IndexService.logger.warn(prefix_name(str), e)
   }
 
+  private def error(str: String) {
+    IndexService.logger.error(prefix_name(str))
+  }
+
   private def error(str: String, e: Throwable) {
     IndexService.logger.error(prefix_name(str), e)
   }
@@ -886,7 +911,7 @@ object IndexService {
           val queryParser = new ClouseauQueryParser(version, "default", analyzer)
           val writerConfig = new IndexWriterConfig(version, analyzer)
           val writer = new IndexWriter(dir, writerConfig)
-          ('ok, node.spawnService[IndexService, IndexServiceArgs](IndexServiceArgs(config, path, queryParser, writer)))
+          ('ok, node.spawnService[IndexService, IndexServiceArgs](IndexServiceArgs(config, rootDir, path, queryParser, writer)))
         case None =>
           ('error, 'no_such_analyzer)
       }