1pub mod go_ahead;
83pub mod world;
84
85use std::sync::{Arc, Condvar, Mutex};
86use std::time::{Duration, Instant};
87
88pub use go_ahead::{GoAhead, GoAheadPolicy, OperatorPolicy};
89#[cfg(feature = "metered")]
90pub use go_ahead::MonetizePolicy;
91pub use world::{Grown, MemberFacts, SurveyFacts, World};
92
93pub const TARGET: &str = "gunnar::grow";
95
96pub const DEFAULT_DRAIN_TIMEOUT: Duration = Duration::from_secs(600);
99pub const DEFAULT_WAIT_MEMBERS_TIMEOUT: Duration = Duration::from_secs(3600);
105pub const UNMOUNTED_TOO_LONG: &str = "grow.unmounted_too_long";
107pub const DRAIN_POLL: Duration = Duration::from_millis(200);
109
110pub fn target_reached(have: u64, asked: u64) -> bool {
115 have >= asked || have.saturating_mul(100) >= asked.saturating_mul(99)
116}
117pub const WAIT_POLL: Duration = Duration::from_secs(2);
119pub const WAIT_POLL_AFTER_TIMEOUT: Duration = Duration::from_secs(15);
121
122#[derive(Debug, Clone, Copy, PartialEq, Eq)]
124pub enum Phase {
125 Idle,
126 Draining,
127 Unmounted,
128 WaitingMembers,
129 Growing,
130 Mounting,
131 Done,
132 Failed,
133}
134
135impl Phase {
136 pub fn as_str(self) -> &'static str {
137 match self {
138 Phase::Idle => "idle",
139 Phase::Draining => "draining",
140 Phase::Unmounted => "unmounted",
141 Phase::WaitingMembers => "waiting-members",
142 Phase::Growing => "growing",
143 Phase::Mounting => "mounting",
144 Phase::Done => "done",
145 Phase::Failed => "failed",
146 }
147 }
148 pub fn is_active(self) -> bool {
150 matches!(
151 self,
152 Phase::Draining | Phase::Unmounted | Phase::WaitingMembers | Phase::Growing | Phase::Mounting
153 )
154 }
155}
156
157#[derive(Debug, Clone, Copy, PartialEq, Eq)]
161pub enum Step {
162 Drain,
163 Unmount,
164 Stop,
165 Wait,
166 Rescan,
167 Reopen,
168 Abort,
169 Fail,
170}
171
172impl Step {
173 pub fn as_str(self) -> &'static str {
174 match self {
175 Step::Drain => "DRAIN",
176 Step::Unmount => "UNMOUNT",
177 Step::Stop => "STOP",
178 Step::Wait => "WAIT",
179 Step::Rescan => "RESCAN",
180 Step::Reopen => "REOPEN",
181 Step::Abort => "ABORT",
182 Step::Fail => "FAIL",
183 }
184 }
185}
186
187#[derive(Debug, Clone, PartialEq, Eq)]
189pub enum Announcement {
190 Draining { kind: &'static str, reason: String },
192 Ready,
194}
195
196#[derive(Debug, Clone, PartialEq, Eq)]
198pub struct Plan {
199 pub target_sectors: u64,
202 pub drain_timeout: Duration,
203 pub wait_timeout: Duration,
204 pub signer: String,
206 pub alone: String,
209}
210
211pub const ALONE_MAX: usize = 600;
213
214pub const ALONE_FLUSH_WAIT: Duration = Duration::from_secs(5);
218
219#[derive(Debug, Clone, Copy, PartialEq, Eq)]
221enum Resume {
222 LookAgain,
223 GiveUp,
224}
225
226#[derive(Debug, Clone, PartialEq, Eq)]
228enum Ended {
229 Eligible,
231 GaveUp,
233 Bound(String),
235}
236
237pub fn carries(survey: &SurveyFacts, target: u64) -> bool {
241 survey.verdict == "complete" && !survey.members.is_empty() && survey.members.iter().all(|m| target_reached(m.set_bytes / 512, target))
242}
243
244#[derive(Debug, Clone)]
246struct State {
247 phase: Phase,
248 since_unix_ms: i64,
249 why: String,
250 detail: String,
251 epoch: u64,
252 target_sectors: u64,
253 signer: String,
254 members: SurveyFacts,
255 identity_in_ram: bool,
256 serving: bool,
257 in_flight: u64,
258 resume: Option<Resume>,
259 wait_timed_out: bool,
261 grown: Option<Grown>,
263 alarm: Option<(String, i64, String)>,
266 alone: String,
269}
270
271impl State {
272 fn idle(serving: bool) -> Self {
273 State {
274 phase: Phase::Idle,
275 since_unix_ms: gunnar_grpc::bus::now_unix_ms(),
276 why: String::new(),
277 detail: String::new(),
278 epoch: 0,
279 target_sectors: 0,
280 signer: String::new(),
281 members: SurveyFacts::default(),
282 identity_in_ram: false,
283 serving,
284 in_flight: 0,
285 resume: None,
286 wait_timed_out: false,
287 grown: None,
288 alarm: None,
289 alone: String::new(),
290 }
291 }
292}
293
294type Announcer = Box<dyn Fn(Announcement) + Send + Sync>;
296
297pub struct GrowMachine {
301 world: Arc<dyn World>,
302 policy: Arc<dyn GoAheadPolicy>,
303 state: Mutex<State>,
304 wake: Condvar,
306 announcer: Mutex<Option<Announcer>>,
307}
308
309impl std::fmt::Debug for GrowMachine {
310 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
311 f.debug_struct("GrowMachine").field("phase", &self.phase()).finish()
312 }
313}
314
315impl GrowMachine {
316 pub fn new(world: Arc<dyn World>) -> Arc<Self> {
318 Self::with_policy(world, Arc::new(OperatorPolicy::default()))
319 }
320
321 pub fn with_policy(world: Arc<dyn World>, policy: Arc<dyn GoAheadPolicy>) -> Arc<Self> {
324 let serving = world.serving();
325 tracing::info!(
328 target: TARGET,
329 event = "grow.up",
330 engine_present = world.engine_present(),
331 serving,
332 "the growth machine is up — idle; {}",
333 if world.engine_present() { "this build grows /data at runtime (Admin.GrowStart)" } else { "no runtime grow engine in this build: the set grows only at a restart" }
334 );
335 Arc::new(GrowMachine {
336 world,
337 policy,
338 state: Mutex::new(State::idle(serving)),
339 wake: Condvar::new(),
340 announcer: Mutex::new(None),
341 })
342 }
343
344 pub fn with_announcer(self: &Arc<Self>, f: impl Fn(Announcement) + Send + Sync + 'static) {
346 *self.announcer.lock().unwrap_or_else(|p| p.into_inner()) = Some(Box::new(f));
347 }
348
349 fn announce(&self, a: Announcement) {
350 match &a {
351 Announcement::Draining { kind, reason } => tracing::warn!(
352 target: TARGET,
353 event = "presence.draining",
354 kind,
355 reason = %reason,
356 "Draining{{kind: {kind}}}: {reason}"
357 ),
358 Announcement::Ready => tracing::info!(target: TARGET, event = "presence.ready", "Ready: /data is back and the gate is open"),
359 }
360 if let Some(f) = self.announcer.lock().unwrap_or_else(|p| p.into_inner()).as_ref() {
361 f(a);
362 }
363 }
364
365 fn lock(&self) -> std::sync::MutexGuard<'_, State> {
366 self.state.lock().unwrap_or_else(|p| p.into_inner())
367 }
368
369 pub fn alarm(&self) -> Option<gunnar_proto::HealthAlarm> {
371 self.lock().alarm.clone().map(|(event, since_unix_ms, detail)| gunnar_proto::HealthAlarm { event, count: 1, since_unix_ms, detail })
372 }
373
374 fn raise(&self, event: &str, detail: String, loud: bool) {
378 let mut s = self.lock();
379 if loud || s.alarm.as_ref().is_none_or(|(e, _, _)| e != event) {
380 tracing::error!(target: TARGET, event, "{detail}");
381 }
382 let since = match &s.alarm {
383 Some((e, since, _)) if e == event => *since,
384 _ => gunnar_grpc::bus::now_unix_ms(),
385 };
386 s.alarm = Some((event.to_owned(), since, detail));
387 }
388
389 pub fn phase(&self) -> Phase {
391 self.lock().phase
392 }
393
394 fn enter(&self, phase: Phase, detail: impl Into<String>) {
395 let mut s = self.lock();
396 let detail = detail.into();
397 tracing::info!(target: TARGET, event = "grow.phase", from = s.phase.as_str(), to = phase.as_str(), "{detail}");
398 s.phase = phase;
399 s.since_unix_ms = gunnar_grpc::bus::now_unix_ms();
400 s.detail = detail;
401 }
402
403 fn note(&self, step: Step) -> Result<(), String> {
404 let (epoch, target) = {
405 let s = self.lock();
406 (s.epoch, s.target_sectors)
407 };
408 self.world.note(epoch, target, step)
409 }
410
411 fn note_failing(&self, step: Step, why: &mut String) {
416 if let Err(e) = self.note(step) {
417 tracing::error!(target: TARGET, event = "grow.journal", step = step.as_str(), %e, "the grow journal did not record a step of a failing growth");
418 why.push_str(&format!("; the grow journal did not record {}: {e}", step.as_str()));
419 }
420 }
421
422 pub fn start(self: &Arc<Self>, request: &gunnar_proto::GrowStartRequest) -> Result<gunnar_proto::GrowStatusResponse, String> {
429 if !self.world.engine_present() {
430 return Err("this gunnar-server was built without the runtime grow engine (feature `grow-runtime`); the set grows only at a restart".into());
431 }
432 if !self.world.hotplug() {
437 return Err(format!(
438 "the running kernel has no PCI hot-plug ({}): a member detached, resized and re-attached while gunnar runs is invisible to it, so a runtime growth is refused — re-image onto a medium whose kernel carries CONFIG_HOTPLUG_PCI, or enlarge every member and restart (the set grows at the next boot)",
439 crate::dashboard::probe::pci_hotplug().evidence
440 ));
441 }
442 if !self.world.journal_present() {
457 return Err(format!(
458 "this box's boot stack keeps no grow journal ({}): korp-start never opened one, so this medium's boot stack does not grow the data set and a runtime growth has nowhere to record the steps a crash would be finished from — serving more needs a fresh install onto a larger set, not another disk. This is the same fact `Admin.Capacity` reports as GROWTH_SEALED",
459 self.world.journal_path().display()
460 ));
461 }
462 let go = GoAhead::parse(&request.go_ahead)?;
463 if go.target_sectors != request.target_sectors_per_member {
464 return Err(format!(
465 "the go-ahead names target_sectors {} and the request {}; they must agree",
466 go.target_sectors, request.target_sectors_per_member
467 ));
468 }
469 let now = gunnar_grpc::bus::now_unix_ms();
470 self.policy.verify(&go, now)?;
471 if !self.world.remember_nonce(&go.nonce)? {
472 return Err(format!("the go-ahead's nonce {:?} was already spent on this box", go.nonce));
473 }
474 let plan = Plan {
475 target_sectors: request.target_sectors_per_member,
476 drain_timeout: if request.drain_timeout_secs == 0 { DEFAULT_DRAIN_TIMEOUT } else { Duration::from_secs(u64::from(request.drain_timeout_secs)) },
477 wait_timeout: if request.wait_members_timeout_secs == 0 { DEFAULT_WAIT_MEMBERS_TIMEOUT } else { Duration::from_secs(u64::from(request.wait_members_timeout_secs)) },
478 signer: go.signer.clone(),
479 alone: {
480 let one_line = go.alone.replace(['\n', '\r'], " ");
481 let mut cut = one_line.trim().to_owned();
482 if cut.len() > ALONE_MAX {
483 let mut at = ALONE_MAX;
484 while !cut.is_char_boundary(at) {
485 at -= 1;
486 }
487 cut.truncate(at);
488 }
489 cut
490 },
491 };
492 {
493 let mut s = self.lock();
494 if s.phase.is_active() {
495 return Err(format!("a growth is already {} (since {}); GrowResume looks again, and a new growth waits for this one", s.phase.as_str(), s.since_unix_ms));
496 }
497 if !s.serving {
498 return Err(format!("/data is not being served ({}); a growth cannot start from here — reboot, or re-attach the members and GrowResume", s.why));
499 }
500 let survey = self.world.survey();
501 if survey.verdict != "complete" {
502 return Err(format!("the data set is not a complete md set right now ({}); nothing to grow", survey.verdict));
503 }
504 if plan.target_sectors > 0 && carries(&survey, plan.target_sectors) {
513 let described = survey.describe();
514 *s = State::idle(true);
515 s.signer = plan.signer.clone();
516 s.target_sectors = plan.target_sectors;
517 s.members = survey;
518 s.phase = Phase::Done;
519 s.since_unix_ms = now;
520 s.detail = format!(
521 "done: every member already carries the {} sectors asked ({described}); nothing was drained and /data never left",
522 plan.target_sectors
523 );
524 drop(s);
525 tracing::warn!(
526 target: TARGET,
527 event = "grow.already_met",
528 signer = %plan.signer,
529 nonce = %go.nonce,
530 target_sectors = plan.target_sectors,
531 "GrowStart for a size the members already carry: done at once, nothing drained or unmounted"
532 );
533 return Ok(self.status());
534 }
535 *s = State::idle(true);
536 s.signer = plan.signer.clone();
537 s.target_sectors = plan.target_sectors;
538 s.members = survey;
539 s.phase = Phase::Draining;
540 s.since_unix_ms = now;
541 s.detail = format!("draining: the gate is paused, waiting up to {} s for {} transfer(s) in flight", plan.drain_timeout.as_secs(), self.world.in_flight());
542 }
543 tracing::warn!(
544 target: TARGET,
545 event = "grow.start",
546 signer = %plan.signer,
547 nonce = %go.nonce,
548 target_sectors = plan.target_sectors,
549 drain_timeout_s = plan.drain_timeout.as_secs(),
550 wait_timeout_s = plan.wait_timeout.as_secs(),
551 "a runtime data-set growth begins"
552 );
553 let started = self.status();
558 let me = Arc::clone(self);
559 std::thread::Builder::new()
560 .name("gunnar-grow".into())
561 .spawn(move || me.run(plan))
562 .map_err(|e| format!("spawning the growth thread: {e}"))?;
563 Ok(started)
564 }
565
566 pub fn resume(&self, give_up: bool) -> Result<gunnar_proto::GrowStatusResponse, String> {
579 if self.lock().phase == Phase::Done {
580 tracing::info!(target: TARGET, event = "grow.resume.done", give_up, "GrowResume on a growth that is already done: nothing to resume");
581 return Ok(self.status());
582 }
583 {
584 let s = self.lock();
585 if matches!(s.phase, Phase::Growing | Phase::Mounting) {
594 drop(s);
595 tracing::info!(target: TARGET, event = "grow.resume.progress", give_up, "GrowResume while the engine runs: the members were already seen; nothing to look at again");
596 return Ok(self.status());
597 }
598 }
599 {
600 let mut s = self.lock();
601 if s.phase != Phase::WaitingMembers && s.phase != Phase::Unmounted {
604 return Err(format!(
605 "GrowResume applies while waiting-members; the growth is {} ({})",
606 s.phase.as_str(),
607 if s.phase == Phase::Idle { "start one with GrowStart" } else { s.detail.as_str() }
608 ));
609 }
610 s.resume = Some(if give_up { Resume::GiveUp } else { Resume::LookAgain });
611 }
612 tracing::info!(target: TARGET, event = "grow.resume", give_up, "GrowResume: looking at the members again");
613 self.wake.notify_all();
614 Ok(self.status())
615 }
616
617 pub fn status(&self) -> gunnar_proto::GrowStatusResponse {
621 let mut s = self.lock().clone();
622 if !s.phase.is_active() && self.world.engine_present() {
623 s.members = self.world.survey();
624 }
625 if s.phase == Phase::Draining {
626 s.in_flight = self.world.in_flight();
627 }
628 let journal = gunnar_grpc::grow_journal::Summary::read(&self.world.journal_path()).map(|j| j.to_wire());
629 let detail = if s.detail.is_empty() {
630 match s.phase {
631 Phase::Idle if !self.world.engine_present() => "no runtime grow engine in this build; the set grows only at a restart".to_owned(),
632 Phase::Idle => format!("idle; members: {}", s.members.describe()),
633 _ => String::new(),
634 }
635 } else {
636 s.detail.clone()
637 };
638 gunnar_proto::GrowStatusResponse {
639 phase: s.phase.as_str().to_owned(),
640 why: s.why.clone(),
641 epoch: s.epoch,
642 target_sectors_per_member: s.target_sectors,
643 members: s
644 .members
645 .members
646 .iter()
647 .map(|m| gunnar_proto::GrowMember {
648 index: m.index,
649 device: m.device.clone(),
650 partition: m.partition.clone(),
651 disk_bytes: m.disk_bytes,
652 set_bytes: m.set_bytes,
653 offered_bytes: m.offered_bytes,
654 larger: m.larger,
655 })
656 .collect(),
657 members_verdict: s.members.verdict.clone(),
658 journal,
659 engine_present: self.world.engine_present(),
660 hotplug: self.world.hotplug(),
661 serving: s.serving,
662 identity_in_ram: s.identity_in_ram,
663 in_flight: s.in_flight,
664 since_unix_ms: s.since_unix_ms,
665 server_unix_ms: gunnar_grpc::bus::now_unix_ms(),
666 detail,
667 go_ahead_signer: s.signer.clone(),
668 set_uuid: s.members.set_uuid.clone(),
669 total_bytes: if s.serving { self.world.mounted_bytes() } else { 0 },
670 }
671 }
672
673 fn run(self: Arc<Self>, plan: Plan) {
676 self.announce(Announcement::Draining { kind: "growing", reason: "the data set is being grown".into() });
679 self.world.pause_admission();
680 let started = Instant::now();
681 loop {
682 let in_flight = self.world.in_flight();
683 self.lock().in_flight = in_flight;
684 let starting = self.world.starting();
688 if in_flight == 0 && starting.is_none() {
689 break;
690 }
691 if started.elapsed() >= plan.drain_timeout {
692 self.world.unpause_admission();
693 let what = match &starting {
694 Some(why) => format!("{in_flight} transfer(s) still in flight and the box still {why}"),
695 None => format!("{in_flight} transfer(s) still in flight"),
696 };
697 self.fail_untouched(format!(
698 "aborted: {what} after the {} s drain; nothing was changed — retry when the box is quieter, or raise drain_timeout_secs",
699 plan.drain_timeout.as_secs()
700 ));
701 return;
702 }
703 std::thread::sleep(DRAIN_POLL);
704 }
705 let flush_wait = if plan.alone.is_empty() { plan.drain_timeout } else { plan.drain_timeout.min(ALONE_FLUSH_WAIT) };
734 let verdict = match self.world.flush_replication(flush_wait) {
735 world::FlushVerdict::Behind { pending, needs_full_resync, why } if !plan.alone.is_empty() => {
736 let said = format!("a standby is behind ({pending} pending, needs_full_resync {needs_full_resync}): {why}");
737 tracing::warn!(target: TARGET, event = "grow.alone", pending, %why, alone = %plan.alone, "the growth goes on ALONE: {said}");
738 world::FlushVerdict::NoStandby(format!("the growth goes on ALONE ({}) — {said}; the twin catches up from the log after the remount", plan.alone))
739 }
740 world::FlushVerdict::CouldNotLook(why) if !plan.alone.is_empty() => {
741 tracing::warn!(target: TARGET, event = "grow.alone", %why, alone = %plan.alone, "the growth goes on ALONE: the standby could not be looked at");
742 world::FlushVerdict::NoStandby(format!("the growth goes on ALONE ({}) — the standby could not be looked at: {why}", plan.alone))
743 }
744 other => other,
745 };
746 if !plan.alone.is_empty() {
747 if let world::FlushVerdict::NoStandby(why) = &verdict {
748 self.lock().alone = why.clone();
749 }
750 }
751 match verdict {
752 world::FlushVerdict::CaughtUp { standbys, waited_ms } => {
753 tracing::info!(target: TARGET, event = "grow.flushed", standbys, waited_ms, "every standby has taken everything; the set may be unmounted");
754 self.lock().detail = format!("drained: nothing in flight and {standbys} standby(s) caught up after {waited_ms} ms");
755 }
756 world::FlushVerdict::NoStandby(why) => {
757 tracing::warn!(target: TARGET, event = "grow.no_standby", %why, "the set is being grown with no standby behind it");
761 self.lock().detail = format!("drained: nothing in flight, and {why}");
762 }
763 world::FlushVerdict::CouldNotLook(why) => {
764 self.world.unpause_admission();
770 self.fail_untouched(format!(
771 "aborted: the gate is shut and nothing is in flight, but whether the twin holds this set COULD NOT BE \
772 ESTABLISHED: {why} \u{2014} nothing was changed. A set unmounted now could only be pulled back from a twin, \
773 and nothing here says there is one that could."
774 ));
775 return;
776 }
777 world::FlushVerdict::Behind { pending, needs_full_resync, why } => {
778 self.world.unpause_admission();
779 self.fail_untouched(format!(
780 "aborted: the gate is shut and nothing is in flight, but a standby is still behind ({pending} pending, needs_full_resync {needs_full_resync}): {why} \u{2014} nothing was changed. Nothing more is coming that would help it, and a set unmounted now could only be pulled back from a twin that does not have it."
781 ));
782 return;
783 }
784 }
785
786 let epoch = self.world.epoch_for(plan.target_sectors);
787 self.lock().epoch = epoch;
788 if let Err(e) = self.note(Step::Drain) {
789 self.world.unpause_admission();
790 self.fail_untouched(format!("aborted: {e}; nothing was changed"));
791 return;
792 }
793
794 self.lock().detail = "draining → unmounting: the identity table is served from RAM, the stores close, /data unmounts, the array stops; the volumes are NOT free yet".to_owned();
807 tracing::info!(target: TARGET, event = "grow.unmounting", "unmounting: identity to RAM, stores close, /data unmounts, the array stops (the phase stays draining until the set is off)");
808 match self.world.suspend_identity("draining for the data-set growth") {
809 Ok(facts) => {
810 self.lock().identity_in_ram = true;
811 tracing::info!(target: TARGET, event = "grow.identity_in_ram", documents = facts.documents, bytes = facts.bytes, "the control store is served from its RAM copy");
812 }
813 Err(e) => {
814 self.world.unpause_admission();
815 let mut why = format!("aborted: the identity table could not be copied into RAM ({e}); nothing was changed");
816 self.note_failing(Step::Abort, &mut why);
817 self.fail_untouched(why);
818 return;
819 }
820 }
821 if let Err(e) = self.world.close_stores() {
822 self.abort_after_suspend(format!("aborted: the stores would not close ({e})"));
823 return;
824 }
825 if let Err(e) = self.world.sync_and_unmount() {
826 self.abort_after_suspend(format!("aborted: {e}"));
827 return;
828 }
829 self.lock().serving = false;
830 if let Err(e) = self.note(Step::Unmount) {
831 self.recover_serve(format!("{e} after the unmount"));
832 return;
833 }
834 if let Err(e) = self.world.stop_array() {
835 self.recover_serve(format!("the array would not stop ({e})"));
836 return;
837 }
838 if let Err(e) = self.note(Step::Stop) {
839 self.recover_serve(format!("{e} after the array stopped"));
840 return;
841 }
842 self.enter(Phase::Unmounted, "unmounted: /data is off, the stores are closed, the array is stopped; the volumes are free");
843
844 if let Err(e) = self.note(Step::Wait) {
847 self.recover_serve(format!("the grow journal did not record WAIT ({e}); the members were not let go of"));
848 return;
849 }
850 self.enter(Phase::WaitingMembers, format!(
851 "waiting for every member to come back larger{}; the outside actor detaches, resizes and re-attaches the volumes; GrowResume looks again",
852 if plan.target_sectors > 0 { format!(" (≥ {} sectors each)", plan.target_sectors) } else { String::new() }
853 ));
854 let waited_from = Instant::now();
855 let ended = loop {
856 let timed_out = waited_from.elapsed() >= plan.wait_timeout;
859 if timed_out {
860 let mut s = self.lock();
861 if !s.wait_timed_out {
862 s.wait_timed_out = true;
863 tracing::warn!(target: TARGET, event = "grow.wait_timeout", waited_s = waited_from.elapsed().as_secs(), "the members did not come back larger within this growth's bound: with every member present /data is mounted back now at the size they carry; with one missing the wait goes on, alarmed — GrowResume looks again, GrowResume{{give_up}} serves what is there");
864 }
865 }
866 let _ = self.world.rescan();
867 let survey = self.world.survey();
868 let eligible = survey.eligible && survey.target_sectors.is_some_and(|t| plan.target_sectors == 0 || target_reached(t, plan.target_sectors));
876 let same_set = survey.set_uuid.is_empty() || {
877 let s = self.lock();
878 s.members.set_uuid.is_empty() || s.members.set_uuid == survey.set_uuid
879 };
880 {
881 let mut s = self.lock();
882 let described = survey.describe();
883 s.members = survey;
884 s.detail = format!(
885 "waiting-members: {described}{}",
886 if !same_set { " — NOT THE SAME SET (uuid differs); refusing to grow it" } else if s.wait_timed_out { " — the wait timeout passed; still waiting for GrowResume or the members" } else { "" }
887 );
888 }
889 if eligible && same_set {
890 break Ended::Eligible;
891 }
892 if timed_out {
903 let waited = waited_from.elapsed().as_secs();
904 let described = self.lock().members.describe();
905 let complete = self.lock().members.verdict == "complete";
906 if complete && same_set {
907 break Ended::Bound(format!(
908 "{UNMOUNTED_TOO_LONG}: /data stayed unmounted {waited} s, past this growth's {} s bound, waiting for members that never came back larger{} ({described}); mounted back at the size the members carry",
909 plan.wait_timeout.as_secs(),
910 if plan.target_sectors > 0 { format!(" (≥ {} sectors each)", plan.target_sectors) } else { String::new() }
911 ));
912 }
913 self.raise(
914 UNMOUNTED_TOO_LONG,
915 format!(
916 "/data has been unmounted {waited} s, past this growth's {} s bound, and cannot be mounted back: {} ({described}) — re-attach the members (GrowResume looks again) or reboot",
917 plan.wait_timeout.as_secs(),
918 if same_set { "a member is missing" } else { "the members are NOT the set that was unmounted" }
919 ),
920 false,
921 );
922 }
923 let resume = self.lock().resume.take();
924 match resume {
925 Some(Resume::GiveUp) => break Ended::GaveUp,
926 Some(Resume::LookAgain) => continue,
927 None => {}
928 }
929 let poll = if timed_out {
933 WAIT_POLL_AFTER_TIMEOUT
934 } else {
935 WAIT_POLL.min(plan.wait_timeout.saturating_sub(waited_from.elapsed()).max(Duration::from_millis(50)))
936 };
937 let guard = self.lock();
938 let (guard, _) = self.wake.wait_timeout(guard, poll).unwrap_or_else(|p| p.into_inner());
939 drop(guard);
940 };
941 if let Ended::Bound(why) = ended {
942 self.raise(UNMOUNTED_TOO_LONG, why.clone(), true);
943 self.recover_serve(why);
944 return;
945 }
946 let give_up = matches!(ended, Ended::GaveUp);
947 if let Err(e) = self.note(Step::Rescan) {
948 tracing::warn!(target: TARGET, event = "grow.journal", %e, "RESCAN not recorded; going on (the disk is the truth)");
949 }
950
951 self.enter(Phase::Growing, if give_up { "growing: the operator gave up waiting; assembling and mounting what is there" } else { "growing: every member is back and larger; the engine runs G1–G6" });
953 match self.world.grow_and_mount() {
954 Ok(grown) => {
955 self.lock().serving = true;
956 self.lock().grown = Some(grown.clone());
957 if grown.grew {
958 if let Err(why) = self.reopen() {
959 let why = format!("the data set grew ({}), but {why}", grown.describe());
960 self.lock().why = why.clone();
961 self.enter(Phase::Failed, format!("failed: {why}"));
962 return;
963 }
964 let alone = std::mem::take(&mut self.lock().alone);
965 self.enter(Phase::Done, if alone.is_empty() { format!("done: {}", grown.describe()) } else { format!("done: {}; grown {alone}", grown.describe()) });
966 tracing::warn!(target: TARGET, event = "grow.done", array_bytes = grown.array_bytes, "the data set grew and is being served");
967 } else {
968 let untouched = give_up && self.lock().members.members.iter().all(|m| !m.larger);
977 let why = if give_up {
978 format!("gave up waiting{}; serving at the size the members share: {}", if untouched { " with no member back larger (nothing was touched)" } else { "" }, grown.describe())
979 } else {
980 format!("the engine did not grow the set; serving at the size the members share: {}", grown.describe())
981 };
982 let mut why = why;
983 self.note_failing(if untouched { Step::Abort } else { Step::Fail }, &mut why);
984 self.lock().why = why.clone();
985 match self.reopen() {
986 Ok(()) => self.enter(Phase::Failed, format!("failed (serving): {why}")),
987 Err(more) => {
988 let why = format!("{why}; and {more}");
989 self.lock().why = why.clone();
990 self.enter(Phase::Failed, format!("failed: {why}"));
991 }
992 }
993 }
994 }
995 Err(e) => {
996 let mut why = format!("FATAL: /data could not be assembled or mounted ({e}); the control plane and GrowStatus keep answering, git is paused, the identity table is served from RAM; re-attach the members and GrowResume is not possible from here — reboot to let the boot path finish the epoch, or reload from the twin (a separate verb)");
1000 self.note_failing(Step::Fail, &mut why);
1001 self.lock().why = why.clone();
1002 self.raise("grow.fatal", why.clone(), true);
1003 self.enter(Phase::Failed, why);
1004 tracing::error!(target: TARGET, event = "grow.fatal", "{e}");
1005 }
1006 }
1007 }
1008
1009 fn fail_untouched(&self, why: String) {
1011 tracing::warn!(target: TARGET, event = "grow.aborted", %why, "the growth aborted with nothing changed");
1012 let mut s = self.lock();
1013 s.why = why.clone();
1014 s.serving = true;
1015 s.identity_in_ram = false;
1016 drop(s);
1017 self.enter(Phase::Failed, why);
1018 }
1019
1020 fn abort_after_suspend(&self, why: String) {
1026 let reopened = self.world.reopen_stores();
1027 match self.world.resume_identity() {
1028 Ok(r) => self.report_resume(&r),
1029 Err(e) => tracing::error!(target: TARGET, event = "grow.identity_resume_failed", %e, "the control store could not be put back on the stripe; it is still served from RAM"),
1030 }
1031 let mut why = why;
1032 self.note_failing(Step::Abort, &mut why);
1033 match reopened {
1034 Ok(()) => {
1035 self.world.unpause_admission();
1036 self.fail_untouched(why);
1037 }
1038 Err(e) => self.stores_stay_closed(format!("{why}; and the stores would not reopen ({e})")),
1039 }
1040 }
1041
1042 fn stores_stay_closed(&self, why: String) {
1045 let why = format!("{why}: git stays paused over closed stores — restart the box to recover");
1046 {
1047 let mut s = self.lock();
1048 s.why = why.clone();
1049 s.serving = false;
1050 }
1051 self.raise("grow.reopen_failed", why.clone(), true);
1052 self.enter(Phase::Failed, why);
1053 }
1054
1055 fn recover_serve(&self, why: String) {
1059 tracing::warn!(target: TARGET, event = "grow.recover", %why, "recovering: mounting what is there and serving at the size it has");
1060 match self.world.grow_and_mount() {
1061 Ok(grown) => {
1062 self.lock().serving = true;
1063 self.lock().grown = Some(grown);
1064 let mut why = why;
1065 self.note_failing(Step::Fail, &mut why);
1066 self.lock().why = why.clone();
1067 match self.reopen() {
1068 Ok(()) => self.enter(Phase::Failed, format!("failed (serving at the old size): {why}")),
1069 Err(more) => {
1070 let why = format!("{why}; and {more}");
1071 self.lock().why = why.clone();
1072 self.enter(Phase::Failed, format!("failed: {why}"));
1073 }
1074 }
1075 }
1076 Err(e) => {
1077 let mut full = format!("FATAL: {why}; and /data could not be mounted back ({e}); the control plane and GrowStatus keep answering, git is paused, the identity table is served from RAM — reboot to let the boot path finish the epoch");
1078 self.note_failing(Step::Fail, &mut full);
1079 self.lock().why = full.clone();
1080 self.raise("grow.fatal", full.clone(), true);
1081 self.enter(Phase::Failed, full);
1082 tracing::error!(target: TARGET, event = "grow.fatal", "{e}");
1083 }
1084 }
1085 }
1086
1087 fn reopen(&self) -> Result<(), String> {
1094 self.enter(Phase::Mounting, "mounting: /data is back; the identity table returns to the stripe and the stores reopen");
1095 match self.world.resume_identity() {
1096 Ok(r) => {
1097 self.lock().identity_in_ram = false;
1098 self.report_resume(&r);
1099 }
1100 Err(e) => tracing::error!(target: TARGET, event = "grow.identity_resume_failed", %e, "the control store could not be put back on the stripe; it is still served from RAM and every write is refused — restart to recover"),
1101 }
1102 if let Err(e) = self.world.reopen_stores() {
1103 let why = format!("the stores did not reopen ({e}): git stays paused over closed stores — restart the box to recover");
1104 self.lock().serving = false;
1105 self.raise("grow.reopen_failed", why.clone(), true);
1106 return Err(why);
1107 }
1108 let noted = self.note(Step::Reopen);
1109 self.world.unpause_admission();
1110 self.announce(Announcement::Ready);
1111 noted.map_err(|e| {
1112 tracing::error!(target: TARGET, event = "grow.journal", step = Step::Reopen.as_str(), %e, "the grow journal did not record REOPEN");
1113 format!("the grow journal did not record REOPEN ({e}); the stores are open and serving")
1114 })
1115 }
1116
1117 fn report_resume(&self, r: &world::ResumeFacts) {
1118 if r.drift.is_empty() {
1119 tracing::info!(target: TARGET, event = "grow.identity_resumed", documents = r.documents_matched, off_stripe_ms = r.off_stripe_ms, "the control store is back on the stripe; the RAM copy and the disk agree");
1120 } else {
1121 tracing::error!(target: TARGET, event = "grow.identity_drift", drift = ?r.drift, "the control store on disk does NOT match the RAM copy that stood in for it — the disk is the truth and is what is served now; these documents changed underneath the growth");
1122 let mut s = self.lock();
1123 s.why = format!("{}{}identity drift on resume: {}", s.why, if s.why.is_empty() { "" } else { "; " }, r.drift.join(", "));
1124 }
1125 }
1126}
1127
1128#[cfg(test)]
1129pub(crate) mod tests;