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()
561 .name("gunnar-grow".into())
562 .spawn(move || me.run(plan))
563 .map_err(|e| format!("spawning the growth thread: {e}"))?;
564 Ok(started)
565 }
566
567 pub fn resume(&self, give_up: bool) -> Result<gunnar_proto::GrowStatusResponse, String> {
580 if self.lock().phase == Phase::Done {
581 tracing::info!(target: TARGET, event = "grow.resume.done", give_up, "GrowResume on a growth that is already done: nothing to resume");
582 return Ok(self.status());
583 }
584 {
585 let s = self.lock();
586 if matches!(s.phase, Phase::Growing | Phase::Mounting) {
595 drop(s);
596 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");
597 return Ok(self.status());
598 }
599 }
600 {
601 let mut s = self.lock();
602 if s.phase != Phase::WaitingMembers && s.phase != Phase::Unmounted {
605 return Err(format!(
606 "GrowResume applies while waiting-members; the growth is {} ({})",
607 s.phase.as_str(),
608 if s.phase == Phase::Idle { "start one with GrowStart" } else { s.detail.as_str() }
609 ));
610 }
611 s.resume = Some(if give_up { Resume::GiveUp } else { Resume::LookAgain });
612 }
613 tracing::info!(target: TARGET, event = "grow.resume", give_up, "GrowResume: looking at the members again");
614 self.wake.notify_all();
615 Ok(self.status())
616 }
617
618 pub fn status(&self) -> gunnar_proto::GrowStatusResponse {
622 let mut s = self.lock().clone();
623 if !s.phase.is_active() && self.world.engine_present() {
624 s.members = self.world.survey();
625 }
626 if s.phase == Phase::Draining {
627 s.in_flight = self.world.in_flight();
628 }
629 let journal = gunnar_grpc::grow_journal::Summary::read(&self.world.journal_path()).map(|j| j.to_wire());
630 let detail = if s.detail.is_empty() {
631 match s.phase {
632 Phase::Idle if !self.world.engine_present() => "no runtime grow engine in this build; the set grows only at a restart".to_owned(),
633 Phase::Idle => format!("idle; members: {}", s.members.describe()),
634 _ => String::new(),
635 }
636 } else {
637 s.detail.clone()
638 };
639 gunnar_proto::GrowStatusResponse {
640 phase: s.phase.as_str().to_owned(),
641 why: s.why.clone(),
642 epoch: s.epoch,
643 target_sectors_per_member: s.target_sectors,
644 members: s
645 .members
646 .members
647 .iter()
648 .map(|m| gunnar_proto::GrowMember {
649 index: m.index,
650 device: m.device.clone(),
651 partition: m.partition.clone(),
652 disk_bytes: m.disk_bytes,
653 set_bytes: m.set_bytes,
654 offered_bytes: m.offered_bytes,
655 larger: m.larger,
656 })
657 .collect(),
658 members_verdict: s.members.verdict.clone(),
659 journal,
660 engine_present: self.world.engine_present(),
661 hotplug: self.world.hotplug(),
662 serving: s.serving,
663 identity_in_ram: s.identity_in_ram,
664 in_flight: s.in_flight,
665 since_unix_ms: s.since_unix_ms,
666 server_unix_ms: gunnar_grpc::bus::now_unix_ms(),
667 detail,
668 go_ahead_signer: s.signer.clone(),
669 set_uuid: s.members.set_uuid.clone(),
670 total_bytes: if s.serving { self.world.mounted_bytes() } else { 0 },
671 }
672 }
673
674 fn run(self: Arc<Self>, plan: Plan) {
677 self.announce(Announcement::Draining { kind: "growing", reason: "the data set is being grown".into() });
680 self.world.pause_admission();
681 let started = Instant::now();
682 loop {
683 let in_flight = self.world.in_flight();
684 self.lock().in_flight = in_flight;
685 let starting = self.world.starting();
689 if in_flight == 0 && starting.is_none() {
690 break;
691 }
692 if started.elapsed() >= plan.drain_timeout {
693 self.world.unpause_admission();
694 let what = match &starting {
695 Some(why) => format!("{in_flight} transfer(s) still in flight and the box still {why}"),
696 None => format!("{in_flight} transfer(s) still in flight"),
697 };
698 self.fail_untouched(format!(
699 "aborted: {what} after the {} s drain; nothing was changed — retry when the box is quieter, or raise drain_timeout_secs",
700 plan.drain_timeout.as_secs()
701 ));
702 return;
703 }
704 std::thread::sleep(DRAIN_POLL);
705 }
706 let flush_wait = if plan.alone.is_empty() { plan.drain_timeout } else { plan.drain_timeout.min(ALONE_FLUSH_WAIT) };
735 let verdict = match self.world.flush_replication(flush_wait) {
736 world::FlushVerdict::Behind { pending, needs_full_resync, why } if !plan.alone.is_empty() => {
737 let said = format!("a standby is behind ({pending} pending, needs_full_resync {needs_full_resync}): {why}");
738 tracing::warn!(target: TARGET, event = "grow.alone", pending, %why, alone = %plan.alone, "the growth goes on ALONE: {said}");
739 world::FlushVerdict::NoStandby(format!("the growth goes on ALONE ({}) — {said}; the twin catches up from the log after the remount", plan.alone))
740 }
741 world::FlushVerdict::CouldNotLook(why) if !plan.alone.is_empty() => {
742 tracing::warn!(target: TARGET, event = "grow.alone", %why, alone = %plan.alone, "the growth goes on ALONE: the standby could not be looked at");
743 world::FlushVerdict::NoStandby(format!("the growth goes on ALONE ({}) — the standby could not be looked at: {why}", plan.alone))
744 }
745 other => other,
746 };
747 if !plan.alone.is_empty() {
748 if let world::FlushVerdict::NoStandby(why) = &verdict {
749 self.lock().alone = why.clone();
750 }
751 }
752 match verdict {
753 world::FlushVerdict::CaughtUp { standbys, waited_ms } => {
754 tracing::info!(target: TARGET, event = "grow.flushed", standbys, waited_ms, "every standby has taken everything; the set may be unmounted");
755 self.lock().detail = format!("drained: nothing in flight and {standbys} standby(s) caught up after {waited_ms} ms");
756 }
757 world::FlushVerdict::NoStandby(why) => {
758 tracing::warn!(target: TARGET, event = "grow.no_standby", %why, "the set is being grown with no standby behind it");
762 self.lock().detail = format!("drained: nothing in flight, and {why}");
763 }
764 world::FlushVerdict::CouldNotLook(why) => {
765 self.world.unpause_admission();
771 self.fail_untouched(format!(
772 "aborted: the gate is shut and nothing is in flight, but whether the twin holds this set COULD NOT BE \
773 ESTABLISHED: {why} \u{2014} nothing was changed. A set unmounted now could only be pulled back from a twin, \
774 and nothing here says there is one that could."
775 ));
776 return;
777 }
778 world::FlushVerdict::Behind { pending, needs_full_resync, why } => {
779 self.world.unpause_admission();
780 self.fail_untouched(format!(
781 "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."
782 ));
783 return;
784 }
785 }
786
787 let epoch = self.world.epoch_for(plan.target_sectors);
788 self.lock().epoch = epoch;
789 if let Err(e) = self.note(Step::Drain) {
790 self.world.unpause_admission();
791 self.fail_untouched(format!("aborted: {e}; nothing was changed"));
792 return;
793 }
794
795 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();
808 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)");
809 match self.world.suspend_identity("draining for the data-set growth") {
810 Ok(facts) => {
811 self.lock().identity_in_ram = true;
812 tracing::info!(target: TARGET, event = "grow.identity_in_ram", documents = facts.documents, bytes = facts.bytes, "the control store is served from its RAM copy");
813 }
814 Err(e) => {
815 self.world.unpause_admission();
816 let mut why = format!("aborted: the identity table could not be copied into RAM ({e}); nothing was changed");
817 self.note_failing(Step::Abort, &mut why);
818 self.fail_untouched(why);
819 return;
820 }
821 }
822 if let Err(e) = self.world.close_stores() {
823 self.abort_after_suspend(format!("aborted: the stores would not close ({e})"));
824 return;
825 }
826 if let Err(e) = self.world.sync_and_unmount() {
827 self.abort_after_suspend(format!("aborted: {e}"));
828 return;
829 }
830 self.lock().serving = false;
831 if let Err(e) = self.note(Step::Unmount) {
832 self.recover_serve(format!("{e} after the unmount"));
833 return;
834 }
835 if let Err(e) = self.world.stop_array() {
836 self.recover_serve(format!("the array would not stop ({e})"));
837 return;
838 }
839 if let Err(e) = self.note(Step::Stop) {
840 self.recover_serve(format!("{e} after the array stopped"));
841 return;
842 }
843 self.enter(Phase::Unmounted, "unmounted: /data is off, the stores are closed, the array is stopped; the volumes are free");
844
845 if let Err(e) = self.note(Step::Wait) {
848 self.recover_serve(format!("the grow journal did not record WAIT ({e}); the members were not let go of"));
849 return;
850 }
851 self.enter(Phase::WaitingMembers, format!(
852 "waiting for every member to come back larger{}; the outside actor detaches, resizes and re-attaches the volumes; GrowResume looks again",
853 if plan.target_sectors > 0 { format!(" (≥ {} sectors each)", plan.target_sectors) } else { String::new() }
854 ));
855 let waited_from = Instant::now();
856 let ended = loop {
857 let timed_out = waited_from.elapsed() >= plan.wait_timeout;
860 if timed_out {
861 let mut s = self.lock();
862 if !s.wait_timed_out {
863 s.wait_timed_out = true;
864 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");
865 }
866 }
867 let _ = self.world.rescan();
868 let survey = self.world.survey();
869 let eligible = survey.eligible && survey.target_sectors.is_some_and(|t| plan.target_sectors == 0 || target_reached(t, plan.target_sectors));
877 let same_set = survey.set_uuid.is_empty() || {
878 let s = self.lock();
879 s.members.set_uuid.is_empty() || s.members.set_uuid == survey.set_uuid
880 };
881 {
882 let mut s = self.lock();
883 let described = survey.describe();
884 s.members = survey;
885 s.detail = format!(
886 "waiting-members: {described}{}",
887 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 { "" }
888 );
889 }
890 if eligible && same_set {
891 break Ended::Eligible;
892 }
893 if timed_out {
904 let waited = waited_from.elapsed().as_secs();
905 let described = self.lock().members.describe();
906 let complete = self.lock().members.verdict == "complete";
907 if complete && same_set {
908 break Ended::Bound(format!(
909 "{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",
910 plan.wait_timeout.as_secs(),
911 if plan.target_sectors > 0 { format!(" (≥ {} sectors each)", plan.target_sectors) } else { String::new() }
912 ));
913 }
914 self.raise(
915 UNMOUNTED_TOO_LONG,
916 format!(
917 "/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",
918 plan.wait_timeout.as_secs(),
919 if same_set { "a member is missing" } else { "the members are NOT the set that was unmounted" }
920 ),
921 false,
922 );
923 }
924 let resume = self.lock().resume.take();
925 match resume {
926 Some(Resume::GiveUp) => break Ended::GaveUp,
927 Some(Resume::LookAgain) => continue,
928 None => {}
929 }
930 let poll = if timed_out {
934 WAIT_POLL_AFTER_TIMEOUT
935 } else {
936 WAIT_POLL.min(plan.wait_timeout.saturating_sub(waited_from.elapsed()).max(Duration::from_millis(50)))
937 };
938 let guard = self.lock();
939 let (guard, _) = self.wake.wait_timeout(guard, poll).unwrap_or_else(|p| p.into_inner());
940 drop(guard);
941 };
942 if let Ended::Bound(why) = ended {
943 self.raise(UNMOUNTED_TOO_LONG, why.clone(), true);
944 self.recover_serve(why);
945 return;
946 }
947 let give_up = matches!(ended, Ended::GaveUp);
948 if let Err(e) = self.note(Step::Rescan) {
949 tracing::warn!(target: TARGET, event = "grow.journal", %e, "RESCAN not recorded; going on (the disk is the truth)");
950 }
951
952 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" });
954 match self.world.grow_and_mount() {
955 Ok(grown) => {
956 self.lock().serving = true;
957 self.lock().grown = Some(grown.clone());
958 if grown.grew {
959 if let Err(why) = self.reopen() {
960 let why = format!("the data set grew ({}), but {why}", grown.describe());
961 self.lock().why = why.clone();
962 self.enter(Phase::Failed, format!("failed: {why}"));
963 return;
964 }
965 let alone = std::mem::take(&mut self.lock().alone);
966 self.enter(Phase::Done, if alone.is_empty() { format!("done: {}", grown.describe()) } else { format!("done: {}; grown {alone}", grown.describe()) });
967 tracing::warn!(target: TARGET, event = "grow.done", array_bytes = grown.array_bytes, "the data set grew and is being served");
968 } else {
969 let untouched = give_up && self.lock().members.members.iter().all(|m| !m.larger);
978 let why = if give_up {
979 format!("gave up waiting{}; serving at the size the members share: {}", if untouched { " with no member back larger (nothing was touched)" } else { "" }, grown.describe())
980 } else {
981 format!("the engine did not grow the set; serving at the size the members share: {}", grown.describe())
982 };
983 let mut why = why;
984 self.note_failing(if untouched { Step::Abort } else { Step::Fail }, &mut why);
985 self.lock().why = why.clone();
986 match self.reopen() {
987 Ok(()) => self.enter(Phase::Failed, format!("failed (serving): {why}")),
988 Err(more) => {
989 let why = format!("{why}; and {more}");
990 self.lock().why = why.clone();
991 self.enter(Phase::Failed, format!("failed: {why}"));
992 }
993 }
994 }
995 }
996 Err(e) => {
997 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)");
1001 self.note_failing(Step::Fail, &mut why);
1002 self.lock().why = why.clone();
1003 self.raise("grow.fatal", why.clone(), true);
1004 self.enter(Phase::Failed, why);
1005 tracing::error!(target: TARGET, event = "grow.fatal", "{e}");
1006 }
1007 }
1008 }
1009
1010 fn fail_untouched(&self, why: String) {
1012 tracing::warn!(target: TARGET, event = "grow.aborted", %why, "the growth aborted with nothing changed");
1013 let mut s = self.lock();
1014 s.why = why.clone();
1015 s.serving = true;
1016 s.identity_in_ram = false;
1017 drop(s);
1018 self.enter(Phase::Failed, why);
1019 }
1020
1021 fn abort_after_suspend(&self, why: String) {
1027 let reopened = self.world.reopen_stores();
1028 match self.world.resume_identity() {
1029 Ok(r) => self.report_resume(&r),
1030 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"),
1031 }
1032 let mut why = why;
1033 self.note_failing(Step::Abort, &mut why);
1034 match reopened {
1035 Ok(()) => {
1036 self.world.unpause_admission();
1037 self.fail_untouched(why);
1038 }
1039 Err(e) => self.stores_stay_closed(format!("{why}; and the stores would not reopen ({e})")),
1040 }
1041 }
1042
1043 fn stores_stay_closed(&self, why: String) {
1046 let why = format!("{why}: git stays paused over closed stores — restart the box to recover");
1047 {
1048 let mut s = self.lock();
1049 s.why = why.clone();
1050 s.serving = false;
1051 }
1052 self.raise("grow.reopen_failed", why.clone(), true);
1053 self.enter(Phase::Failed, why);
1054 }
1055
1056 fn recover_serve(&self, why: String) {
1060 tracing::warn!(target: TARGET, event = "grow.recover", %why, "recovering: mounting what is there and serving at the size it has");
1061 match self.world.grow_and_mount() {
1062 Ok(grown) => {
1063 self.lock().serving = true;
1064 self.lock().grown = Some(grown);
1065 let mut why = why;
1066 self.note_failing(Step::Fail, &mut why);
1067 self.lock().why = why.clone();
1068 match self.reopen() {
1069 Ok(()) => self.enter(Phase::Failed, format!("failed (serving at the old size): {why}")),
1070 Err(more) => {
1071 let why = format!("{why}; and {more}");
1072 self.lock().why = why.clone();
1073 self.enter(Phase::Failed, format!("failed: {why}"));
1074 }
1075 }
1076 }
1077 Err(e) => {
1078 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");
1079 self.note_failing(Step::Fail, &mut full);
1080 self.lock().why = full.clone();
1081 self.raise("grow.fatal", full.clone(), true);
1082 self.enter(Phase::Failed, full);
1083 tracing::error!(target: TARGET, event = "grow.fatal", "{e}");
1084 }
1085 }
1086 }
1087
1088 fn reopen(&self) -> Result<(), String> {
1095 self.enter(Phase::Mounting, "mounting: /data is back; the identity table returns to the stripe and the stores reopen");
1096 match self.world.resume_identity() {
1097 Ok(r) => {
1098 self.lock().identity_in_ram = false;
1099 self.report_resume(&r);
1100 }
1101 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"),
1102 }
1103 if let Err(e) = self.world.reopen_stores() {
1104 let why = format!("the stores did not reopen ({e}): git stays paused over closed stores — restart the box to recover");
1105 self.lock().serving = false;
1106 self.raise("grow.reopen_failed", why.clone(), true);
1107 return Err(why);
1108 }
1109 let noted = self.note(Step::Reopen);
1110 self.world.unpause_admission();
1111 self.announce(Announcement::Ready);
1112 noted.map_err(|e| {
1113 tracing::error!(target: TARGET, event = "grow.journal", step = Step::Reopen.as_str(), %e, "the grow journal did not record REOPEN");
1114 format!("the grow journal did not record REOPEN ({e}); the stores are open and serving")
1115 })
1116 }
1117
1118 fn report_resume(&self, r: &world::ResumeFacts) {
1119 if r.drift.is_empty() {
1120 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");
1121 } else {
1122 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");
1123 let mut s = self.lock();
1124 s.why = format!("{}{}identity drift on resume: {}", s.why, if s.why.is_empty() { "" } else { "; " }, r.drift.join(", "));
1125 }
1126 }
1127}
1128
1129#[cfg(test)]
1130pub(crate) mod tests;