fix(build): a record being published is waited out, and only an ended process is a lost controller

On Windows the name a rename installs can be refused to a reader for a
short interval. The waiter read that refusal as an unreadable controller
record and declared a running compiler lost: 10 of 40 Windows-host cross
builds failed this way (#890). Records and statuses are now opened with a
bounded backoff (kRecordSettle, 250 ms) while the name exists; an absent
name is absent at once and a refusal that outlasts the bound is an error.
The controller's state is three-valued: only an identity that names an
ended process is ended; an unreadable record is unknown and the waiter
keeps waiting within its existing bounds. A loss is reported with the
observation that decided it. copy_atomic publishes through the host
namespace like every other record of the module, so a name in Windows'
long spelling stays valid. A cancelled managed supervisor writes its
status, and a registration failure reaches the supervisor log.

Co-authored-by: speak-agent <248744407+speak-agent@users.noreply.github.com>

Author: Sunrisepeak

src/build/schedule/compiler_transaction.cppm

@@ -40,18 +40,38 @@ std::string pack(std::string_view tag, const Fields& fields) {
 // Atomic visibility across process loss; storage durability is not promised.
 bool replace_record(const fs::path& path,std::string_view bytes,std::string* diagnostic=nullptr);
 bool publish(const fs::path& path,std::string_view tag,const Fields& fields);
-Record read(const fs::path& path, std::string_view tag) {
-    std::ifstream input(path,std::ios::binary);
-    if (!input) {
+// A RECORD THAT EXISTS AND CANNOT YET BE OPENED IS BEING PUBLISHED. On Windows
+// the name a rename installs can be refused to a reader for a short interval
+// (the rename's own handle, or a scanner holding the new file); a reader that
+// took that refusal as the record's content declared a live compiler lost
+// (mcpp-community/mcpp#889, measured 10 of 40 Windows cross builds). The open
+// is retried with backoff for at most kRecordSettle; a name that is absent is
+// absent at once, and a refusal that outlasts the bound is still an error.
+inline constexpr auto kRecordSettle=std::chrono::milliseconds(250);
+enum class Opened { yes, absent, refused };
+// Opens `path' for reading, waiting out a publication in progress.
+Opened open_settled(std::ifstream& input,const fs::path& path) {
+    input.open(path,std::ios::binary);
+    if(input)return Opened::yes;
+    const auto limit=std::chrono::steady_clock::now()+kRecordSettle;
+    auto pause=std::chrono::milliseconds(1);
+    for(;;) {
         std::error_code ec;
-        if (!fs::exists(path,ec) && !ec) return std::optional<Fields>{};
-        if(ec)return std::unexpected("cannot read compiler transaction record");
-        // An atomic publisher may install the record between the failed open
-        // and the existence probe. Retry once; a persistent refusal still
-        // fails rather than being mistaken for an absent record.
+        const bool present=fs::exists(path,ec);
+        if(!present&&!ec)return Opened::absent;
+        input.clear();
         input.open(path,std::ios::binary);
-        if(!input)return std::unexpected("cannot read compiler transaction record");
+        if(input)return Opened::yes;
+        if(std::chrono::steady_clock::now()>=limit)return Opened::refused;
+        std::this_thread::sleep_for(pause);
+        pause=std::min(pause*2,std::chrono::milliseconds(32));
     }
+}
+Record read(const fs::path& path, std::string_view tag) {
+    std::ifstream input;
+    const auto opened=open_settled(input,path);
+    if(opened==Opened::absent)return std::optional<Fields>{};
+    if(opened==Opened::refused)return std::unexpected("cannot read compiler transaction record");
     std::string bytes(65537,'\0');
     input.read(bytes.data(),bytes.size());bytes.resize(static_cast<std::size_t>(input.gcount()));
     if(input.bad())return std::unexpected("cannot read compiler transaction record");
@@ -118,7 +138,11 @@ bool copy_atomic(const fs::path& from,const fs::path& to) {
     fs::copy_file(from,temp,fs::copy_options::overwrite_existing,ec);if(ec)return false;
     const auto time=fs::last_write_time(from,ec);if(ec)return false;
     fs::last_write_time(temp,time,ec);if(ec)return false;
-    return bool(os::fs::rename(temp,to,true));
+    // The host namespace, as for every record of this module (the copy was
+    // written through it); a name in Windows' long spelling stays valid.
+    fs::rename(temp,to,ec);
+    if(ec) {fs::remove(temp,ec);return false;}
+    return true;
 }
 struct Attempt {
     fs::path directory, bmi, object, slot, dep;
@@ -299,17 +323,26 @@ bool release_reservation(const fs::path& reservation,const fs::path& attempt) {
     if((**current)[0]!=spelling(attempt))return true; // a later, separately locked slot
     std::error_code ec;fs::remove(reservation,ec);return !ec;
 }
-bool controller_alive(const Attempt& attempt) {
+// WHETHER A CONTROLLER IS RUNNING IS A QUESTION ABOUT A PROCESS, NOT ABOUT A
+// FILE. Only an identity that names an ended process answers `ended'; a record
+// that cannot be read yet answers `unknown', and the caller keeps waiting
+// within its own bounds. A record that is absent is the existing startup case.
+enum class Controller { running, ended, unknown };
+Controller controller_state(const Attempt& attempt) {
     auto controller=read(attempt.directory/"controller","mcpp-controller-v1");
-    if(!controller)return false;
-    if(!*controller)return true; // the existing startup bound applies
-    if((**controller).size()!=3)return false;
+    if(!controller)return Controller::unknown;
+    if(!*controller)return Controller::running; // the existing startup bound applies
+    if((**controller).size()!=3)return Controller::ended;
     os::process::program who{};const auto& fields=**controller;
     const auto id=std::from_chars(fields[0].data(),fields[0].data()+fields[0].size(),who.id);
     const auto birth=std::from_chars(fields[1].data(),fields[1].data()+fields[1].size(),who.started_ns);
-    return id.ec==std::errc{}&&id.ptr==fields[0].data()+fields[0].size()
-        &&birth.ec==std::errc{}&&birth.ptr==fields[1].data()+fields[1].size()
-        &&who.id&&who.started_ns&&os::process::alive(who);
+    const bool valid=id.ec==std::errc{}&&id.ptr==fields[0].data()+fields[0].size()
+        &&birth.ec==std::errc{}&&birth.ptr==fields[1].data()+fields[1].size()&&who.id&&who.started_ns;
+    if(!valid)return Controller::ended;
+    return os::process::alive(who) ? Controller::running : Controller::ended;
+}
+std::string_view describe(Controller c) {
+    return c==Controller::running ? "running" : c==Controller::ended ? "ended" : "unknown";
 }
 
 std::expected<std::string,std::string> receive(os::ipc::connection& connection,std::chrono::milliseconds bound) {
@@ -399,7 +432,9 @@ private:
             if(*retired) {it=held_.erase(it);continue;}
             auto attempt=Attempt::load(directory);
             if(!attempt) {fail(attempt.error());++it;continue;}
-            if(!controller_alive(*attempt)) {
+            const auto state=controller_state(*attempt);
+            if(state==Controller::unknown) {++it;continue;}  // asked again at the next audit
+            if(state==Controller::ended) {
                 // Retirement can be published between the first read and the
                 // controller's exit. Observe it again before reporting loss.
                 retired=read(directory/"retired","mcpp-retired-v1");
@@ -430,7 +465,9 @@ private:
             const auto actor=named((**unit)[3]);
             if(actor.parent_path()!=generation_/"actors") {send(connection,"REFUSED");continue;}
             auto lease=os::fs::lock::acquire(actor,os::fs::lock::kind::exclusive,false);
-            if(lease||lease.error().code!=os::errc::again||!controller_alive(*prepared)) {send(connection,"REFUSED");continue;}
+            // The actor's lock is held, so its controller is running; a record
+            // still settling does not refuse the registration.
+            if(lease||lease.error().code!=os::errc::again||controller_state(*prepared)==Controller::ended) {send(connection,"REFUSED");continue;}
             auto retained=os::process::unit_observer::open((**unit)[0]);
             if(!retained) {send(connection,"REFUSED");continue;}
             std::lock_guard guard(admission_);

src/build/schedule/detach_codegen.cppm

@@ -299,8 +299,8 @@ void withdraw_outputs(const std::filesystem::path& slot,
 // the supervisor's status is lost, and the unit is taken as failed rather than
 // waited for until the hard limit.
 std::optional<int> read_rc(const std::filesystem::path& slot) {
-    std::ifstream in(suffixed(slot, ".rc"));
-    if (!in) return std::nullopt;
+    std::ifstream in;
+    if (transactions::open_settled(in, suffixed(slot, ".rc")) != transactions::Opened::yes) return std::nullopt;
     int rc = 0;
     if (!(in >> rc)) {
         std::println(std::cerr, "mcpp: the status of the compiler for {} was not written "
@@ -548,6 +548,15 @@ void record_supervisor_error(const std::filesystem::path& slot,std::string_view
     if(log)log << "mcpp: " << reason << '\n';
 }
 
+// What the waiter observed when it concluded that the controller is gone: one
+// line, without a path, so a failure states which observation decided it.
+void report_controller_loss(const std::filesystem::path& slot,transactions::Controller controller,bool supervising) {
+    std::error_code ec;
+    const bool log=std::filesystem::exists(suffixed(slot,".log"),ec);
+    std::println(std::cerr,"mcpp: compiler controller observation: controller {}, supervisor {}, status absent, log {}",
+                 transactions::describe(controller),supervising ? "running" : "ended",log ? "present" : "absent");
+}
+
 void replay_supervisor_log(const std::filesystem::path& slot) {
     std::ifstream log(suffixed(slot,".log"));
     if(log)std::cerr << log.rdbuf();
@@ -827,12 +836,14 @@ int compile_release_at_bmi(const CompileRequest& req) {
         }
         if(attempt && clock::now()>=nextControllerCheck) {
             nextControllerCheck=clock::now()+std::chrono::milliseconds(50);
-            if(!transactions::controller_alive(*attempt)
-                ||(supervisor&&!mcpplibs::os::process::alive(*supervisor))) {
+            const auto controller=transactions::controller_state(*attempt);
+            const bool supervising=!supervisor||mcpplibs::os::process::alive(*supervisor);
+            if(controller==transactions::Controller::ended||!supervising) {
                 // The final status can be published between the first read and
                 // this liveness observation. The next iteration validates it.
                 if(read_rc(actual.slot))continue;
                 replay_supervisor_log(actual.slot);
+                report_controller_loss(actual.slot,controller,supervising);
                 std::println(std::cerr,"mcpp: compiler controller ended without a final status; retaining its outputs until unit recovery");
                 return 1;
             }
@@ -885,9 +896,10 @@ int supervise(const std::filesystem::path& slot,
         auto unit=std::make_shared<mcpplibs::os::process::owned_unit>(std::move(*created));
         InterruptCleanup cleanup{mcpp::run::add_cleanup([actor,unit] {unit->terminate();})};
         if(auto registered=attempt.register_unit(*unit,actor->path,permit->reservation);!registered) {
-            std::println(std::cerr,"mcpp: {}",registered.error());(void)write_rc(slot,2);return 2;
+            record_supervisor_error(slot,registered.error());(void)write_rc(slot,2);return 2;
         }
-        if(!actor->owner.active())return 130;
+        // Every exit leaves a status, a cancelled one included.
+        if(!actor->owner.active()) {(void)write_rc(slot,130);return 130;}
         if(!transactions::publish(attempt.directory/"admitted","mcpp-admitted-v1",{})) {
             (void)write_rc(slot,2);return 2;
         }
@@ -982,11 +994,13 @@ int await_unit(const std::filesystem::path& slot, const std::filesystem::path& o
         if (!actor.owner.active()) return 130;
         if(attempt&&clock::now()>=nextControllerCheck) {
             nextControllerCheck=clock::now()+std::chrono::milliseconds(50);
-            if(!transactions::controller_alive(*attempt)) {
+            const auto controller=transactions::controller_state(*attempt);
+            if(controller==transactions::Controller::ended) {
                 // A controller may publish its status and exit after the read
                 // in the loop condition. Let the normal result path inspect it.
                 if((rc=read_rc(slot)))break;
                 replay_supervisor_log(slot);
+                report_controller_loss(slot,controller,true);
                 std::println(std::cerr,"mcpp: compiler controller ended without a status; unit recovery will retain or invalidate its outputs");
                 return 1;
             }

tests/e2e/helpers/record_open_refusal.c

@@ -0,0 +1,32 @@
+#define _GNU_SOURCE
+#include <dlfcn.h>
+#include <errno.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+/* Refuses opening an existing record for reading, as Windows refuses a name a
+ * rename has just installed: MCPP_TEST_REFUSE_NAME names the file and
+ * MCPP_TEST_REFUSE_COUNT how many opens are refused (-1: every one). Each
+ * refusal appends a line to `<name>-refused`. */
+static FILE *(*native_open)(const char *, const char *);
+static long refused;
+
+FILE *fopen64(const char *path, const char *mode) {
+    if (!native_open) native_open = dlsym(RTLD_NEXT, "fopen64");
+    const char *name = getenv("MCPP_TEST_REFUSE_NAME");
+    const char *count = getenv("MCPP_TEST_REFUSE_COUNT");
+    const long limit = count ? strtol(count, NULL, 10) : 0;
+    if (name && strcmp(path, name) == 0 && mode[0] == 'r' && (limit < 0 || refused < limit)) {
+        ++refused;
+        char marker[512];
+        snprintf(marker, sizeof marker, "%s-refused", name);
+        FILE *m = native_open(marker, "ab");
+        if (m) { fputs("refused\n", m); fclose(m); }
+        errno = EACCES;
+        return NULL;
+    }
+    return native_open(path, mode);
+}
+
+FILE *fopen(const char *path, const char *mode) { return fopen64(path, mode); }

tests/unit/test_compiler_transaction.cpp

@@ -288,6 +288,84 @@ TEST(CompilerTransaction, RecordAppearanceBetweenOpenAndExistsIsNotCorruption) {
     EXPECT_EQ(result.exit_code,0)<<result.output;
     EXPECT_TRUE(fs::exists(t.root/"record-open-interposed"));
 }
+TEST(CompilerTransaction, RecordRefusalDriver) {
+    if(!mcpp::invocation::get("MCPP_TEST_RECORD_REFUSAL_DRIVER"))GTEST_SKIP()<<"subprocess driver";
+    const auto started=std::chrono::steady_clock::now();
+    auto record=tx::read("record","schema");
+    const auto elapsed=std::chrono::steady_clock::now()-started;
+    const bool forever=mcpp::invocation::get("MCPP_TEST_REFUSE_COUNT").value_or("")=="-1";
+    if(forever) {
+        EXPECT_FALSE(record);
+        EXPECT_LT(elapsed,std::chrono::seconds(2));
+    } else {
+        ASSERT_TRUE(record&&*record);
+        EXPECT_EQ((**record)[0],"published");
+    }
+}
+TEST(CompilerTransaction, ControllerRefusalDriver) {
+    if(!mcpp::invocation::get("MCPP_TEST_CONTROLLER_REFUSAL_DRIVER"))GTEST_SKIP()<<"subprocess driver";
+    auto attempt=tx::Attempt::load(fs::path(*mcpp::invocation::get("MCPP_TEST_ATTEMPT")));
+    ASSERT_TRUE(attempt)<<attempt.error();
+    EXPECT_EQ(tx::controller_state(*attempt),tx::Controller::unknown);
+}
+namespace {
+// Runs one driver of this binary with the refusal interposer preloaded.
+std::optional<mcpp::run::RunResult> refused_driver(const fs::path& root,std::string_view filter,
+                                               std::vector<std::pair<std::string,std::string>> env) {
+    const auto library=root/"record-open-refusal.so";
+    const auto source=fs::current_path()/"tests/e2e/helpers/record_open_refusal.c";
+    auto compiled=mcpp::run::execute({"/usr/bin/cc","-shared","-fPIC","-Wall","-Wextra","-Werror",
+        source.string(),"-o",library.string(),"-ldl"},{.total=std::chrono::seconds(10)});
+    if(compiled.exit_code!=0) {ADD_FAILURE()<<compiled.output;return std::nullopt;}
+    env.push_back({"LD_PRELOAD",library.string()});
+    return mcpp::run::execute({mcpp::invocation::self_exe_path().string(),
+        std::format("--gtest_filter=CompilerTransaction.{}",filter)},
+        {.env=env,.cwd=root.string(),.total=std::chrono::seconds(20)});
+}
+bool preloadable() {
+    if constexpr (!mcpplibs::os::is_linux)return false;
+    const auto runtime=mcpp::runtime::elf::inspect_elf_runtime(mcpp::invocation::self_exe_path());
+    return runtime && runtime->interp.find("ld-linux")!=std::string::npos;
+}
+std::size_t refusals(const fs::path& marker) {
+    std::ifstream in(marker);std::size_t n=0;std::string line;
+    while(std::getline(in,line))++n;
+    return n;
+}
+}  // namespace
+TEST(CompilerTransaction, RecordRefusedWhilePublishedIsWaitedOut) {
+    if(!preloadable())GTEST_SKIP()<<"open interposition requires Linux and the dynamic glibc loader";
+    Transaction t;
+    ASSERT_TRUE(tx::publish(t.root/"record","schema",{"published"}));
+    const auto result=refused_driver(t.root,"RecordRefusalDriver",
+        {{"MCPP_TEST_RECORD_REFUSAL_DRIVER","1"},{"MCPP_TEST_REFUSE_NAME","record"},{"MCPP_TEST_REFUSE_COUNT","4"}});
+    ASSERT_TRUE(result);
+    EXPECT_EQ(result->exit_code,0)<<result->output;
+    // Four refusals: the reader before 2026.11.7.1 retried once and failed.
+    EXPECT_EQ(refusals(t.root/"record-refused"),4u);
+}
+TEST(CompilerTransaction, PersistentRefusalIsABoundedError) {
+    if(!preloadable())GTEST_SKIP()<<"open interposition requires Linux and the dynamic glibc loader";
+    Transaction t;
+    ASSERT_TRUE(tx::publish(t.root/"record","schema",{"published"}));
+    const auto result=refused_driver(t.root,"RecordRefusalDriver",
+        {{"MCPP_TEST_RECORD_REFUSAL_DRIVER","1"},{"MCPP_TEST_REFUSE_NAME","record"},{"MCPP_TEST_REFUSE_COUNT","-1"}});
+    ASSERT_TRUE(result);
+    EXPECT_EQ(result->exit_code,0)<<result->output;
+    EXPECT_GE(refusals(t.root/"record-refused"),2u);
+}
+TEST(CompilerTransaction, UnreadableControllerRecordIsNotAnEndedController) {
+    if(!preloadable())GTEST_SKIP()<<"open interposition requires Linux and the dynamic glibc loader";
+    Transaction t;
+    auto a=t.prepare();
+    ASSERT_TRUE(tx::publish(a.directory/"controller","mcpp-controller-v1",{"1","1",tx::spelling(t.generation/"actors/helper.lock")}));
+    const auto result=refused_driver(t.root,"ControllerRefusalDriver",
+        {{"MCPP_TEST_CONTROLLER_REFUSAL_DRIVER","1"},{"MCPP_TEST_ATTEMPT",a.directory.string()},
+         {"MCPP_TEST_REFUSE_NAME",(a.directory/"controller").string()},
+         {"MCPP_TEST_REFUSE_COUNT","-1"}});
+    ASSERT_TRUE(result);
+    EXPECT_EQ(result->exit_code,0)<<result->output;
+}
 TEST(CompilerTransaction, FailedCloseCannotPublishOrAcknowledgeARecord) {
     if constexpr (!mcpplibs::os::is_linux)GTEST_SKIP()<<"Linux libc close failure control";
     const auto runtime=mcpp::runtime::elf::inspect_elf_runtime(mcpp::invocation::self_exe_path());