1pub mod go_ahead;
83pub mod world;
84
85use std::sync::{Arc, Condvar, Mutex};
86use std::time::{Duration, Instant};
87
88#[cfg(feature = "metered")]
89pub use go_ahead::MonetizePolicy;
90pub use go_ahead::{GoAhead, GoAheadPolicy, OperatorPolicy};
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
153 | Phase::Unmounted
154 | Phase::WaitingMembers
155 | Phase::Growing
156 | Phase::Mounting
157 )
158 }
159}
160
161#[derive(Debug, Clone, Copy, PartialEq, Eq)]
165pub enum Step {
166 Drain,
167 Unmount,
168 Stop,
169 Wait,
170 Rescan,
171 Reopen,
172 Abort,
173 Fail,
174}
175
176impl Step {
177 pub fn as_str(self) -> &'static str {
178 match self {
179 Step::Drain => "DRAIN",
180 Step::Unmount => "UNMOUNT",
181 Step::Stop => "STOP",
182 Step::Wait => "WAIT",
183 Step::Rescan => "RESCAN",
184 Step::Reopen => "REOPEN",
185 Step::Abort => "ABORT",
186 Step::Fail => "FAIL",
187 }
188 }
189}
190
191#[derive(Debug, Clone, PartialEq, Eq)]
193pub enum Announcement {
194 Draining { kind: &'static str, reason: String },
196 Ready,
198}
199
200#[derive(Debug, Clone, PartialEq, Eq)]
202pub struct Plan {
203 pub target_sectors: u64,
206 pub drain_timeout: Duration,
207 pub wait_timeout: Duration,
208 pub signer: String,
210 pub alone: String,
213}
214
215pub const ALONE_MAX: usize = 600;
217
218pub const ALONE_FLUSH_WAIT: Duration = Duration::from_secs(5);
222
223#[derive(Debug, Clone, Copy, PartialEq, Eq)]
225enum Resume {
226 LookAgain,
227 GiveUp,
228}
229
230#[derive(Debug, Clone, PartialEq, Eq)]
232enum Ended {
233 Eligible,
235 GaveUp,
237 Bound(String),
239}
240
241pub fn carries(survey: &SurveyFacts, target: u64) -> bool {
245 survey.verdict == "complete"
246 && !survey.members.is_empty()
247 && survey
248 .members
249 .iter()
250 .all(|m| target_reached(m.set_bytes / 512, target))
251}
252
253#[derive(Debug, Clone)]
255struct State {
256 phase: Phase,
257 since_unix_ms: i64,
258 why: String,
259 detail: String,
260 epoch: u64,
261 target_sectors: u64,
262 signer: String,
263 members: SurveyFacts,
264 identity_in_ram: bool,
265 serving: bool,
266 in_flight: u64,
267 resume: Option<Resume>,
268 wait_timed_out: bool,
270 grown: Option<Grown>,
272 alarm: Option<(String, i64, String)>,
275 alone: String,
278}
279
280impl State {
281 fn idle(serving: bool) -> Self {
282 State {
283 phase: Phase::Idle,
284 since_unix_ms: gunnar_grpc::bus::now_unix_ms(),
285 why: String::new(),
286 detail: String::new(),
287 epoch: 0,
288 target_sectors: 0,
289 signer: String::new(),
290 members: SurveyFacts::default(),
291 identity_in_ram: false,
292 serving,
293 in_flight: 0,
294 resume: None,
295 wait_timed_out: false,
296 grown: None,
297 alarm: None,
298 alone: String::new(),
299 }
300 }
301}
302
303type Announcer = Box<dyn Fn(Announcement) + Send + Sync>;
305
306pub struct GrowMachine {
310 world: Arc<dyn World>,
311 policy: Arc<dyn GoAheadPolicy>,
312 state: Mutex<State>,
313 wake: Condvar,
315 announcer: Mutex<Option<Announcer>>,
316}
317
318impl std::fmt::Debug for GrowMachine {
319 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
320 f.debug_struct("GrowMachine")
321 .field("phase", &self.phase())
322 .finish()
323 }
324}
325
326impl GrowMachine {
327 pub fn new(world: Arc<dyn World>) -> Arc<Self> {
329 Self::with_policy(world, Arc::new(OperatorPolicy::default()))
330 }
331
332 pub fn with_policy(world: Arc<dyn World>, policy: Arc<dyn GoAheadPolicy>) -> Arc<Self> {
335 let serving = world.serving();
336 tracing::info!(
339 target: TARGET,
340 event = "grow.up",
341 engine_present = world.engine_present(),
342 serving,
343 "the growth machine is up — idle; {}",
344 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" }
345 );
346 Arc::new(GrowMachine {
347 world,
348 policy,
349 state: Mutex::new(State::idle(serving)),
350 wake: Condvar::new(),
351 announcer: Mutex::new(None),
352 })
353 }
354
355 pub fn with_announcer(self: &Arc<Self>, f: impl Fn(Announcement) + Send + Sync + 'static) {
357 *self.announcer.lock().unwrap_or_else(|p| p.into_inner()) = Some(Box::new(f));
358 }
359
360 fn announce(&self, a: Announcement) {
361 match &a {
362 Announcement::Draining { kind, reason } => tracing::warn!(
363 target: TARGET,
364 event = "presence.draining",
365 kind,
366 reason = %reason,
367 "Draining{{kind: {kind}}}: {reason}"
368 ),
369 Announcement::Ready => {
370 tracing::info!(target: TARGET, event = "presence.ready", "Ready: /data is back and the gate is open")
371 }
372 }
373 if let Some(f) = self
374 .announcer
375 .lock()
376 .unwrap_or_else(|p| p.into_inner())
377 .as_ref()
378 {
379 f(a);
380 }
381 }
382
383 fn lock(&self) -> std::sync::MutexGuard<'_, State> {
384 self.state.lock().unwrap_or_else(|p| p.into_inner())
385 }
386
387 pub fn alarm(&self) -> Option<gunnar_proto::HealthAlarm> {
389 self.lock()
390 .alarm
391 .clone()
392 .map(|(event, since_unix_ms, detail)| gunnar_proto::HealthAlarm {
393 event,
394 count: 1,
395 since_unix_ms,
396 detail,
397 })
398 }
399
400 fn raise(&self, event: &str, detail: String, loud: bool) {
404 let mut s = self.lock();
405 if loud || s.alarm.as_ref().is_none_or(|(e, _, _)| e != event) {
406 tracing::error!(target: TARGET, event, "{detail}");
407 }
408 let since = match &s.alarm {
409 Some((e, since, _)) if e == event => *since,
410 _ => gunnar_grpc::bus::now_unix_ms(),
411 };
412 s.alarm = Some((event.to_owned(), since, detail));
413 }
414
415 pub fn phase(&self) -> Phase {
417 self.lock().phase
418 }
419
420 fn enter(&self, phase: Phase, detail: impl Into<String>) {
421 let mut s = self.lock();
422 let detail = detail.into();
423 tracing::info!(target: TARGET, event = "grow.phase", from = s.phase.as_str(), to = phase.as_str(), "{detail}");
424 s.phase = phase;
425 s.since_unix_ms = gunnar_grpc::bus::now_unix_ms();
426 s.detail = detail;
427 }
428
429 fn note(&self, step: Step) -> Result<(), String> {
430 let (epoch, target) = {
431 let s = self.lock();
432 (s.epoch, s.target_sectors)
433 };
434 self.world.note(epoch, target, step)
435 }
436
437 fn note_failing(&self, step: Step, why: &mut String) {
442 if let Err(e) = self.note(step) {
443 tracing::error!(target: TARGET, event = "grow.journal", step = step.as_str(), %e, "the grow journal did not record a step of a failing growth");
444 why.push_str(&format!(
445 "; the grow journal did not record {}: {e}",
446 step.as_str()
447 ));
448 }
449 }
450
451 pub fn start(
458 self: &Arc<Self>,
459 request: &gunnar_proto::GrowStartRequest,
460 ) -> Result<gunnar_proto::GrowStatusResponse, String> {
461 if !self.world.engine_present() {
462 return Err("this gunnar-server was built without the runtime grow engine (feature `grow-runtime`); the set grows only at a restart".into());
463 }
464 if !self.world.hotplug() {
469 return Err(format!(
470 "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)",
471 crate::dashboard::probe::pci_hotplug().evidence
472 ));
473 }
474 if !self.world.journal_present() {
489 return Err(format!(
490 "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",
491 self.world.journal_path().display()
492 ));
493 }
494 let go = GoAhead::parse(&request.go_ahead)?;
495 if go.target_sectors != request.target_sectors_per_member {
496 return Err(format!(
497 "the go-ahead names target_sectors {} and the request {}; they must agree",
498 go.target_sectors, request.target_sectors_per_member
499 ));
500 }
501 let now = gunnar_grpc::bus::now_unix_ms();
502 self.policy.verify(&go, now)?;
503 if !self.world.remember_nonce(&go.nonce)? {
504 return Err(format!(
505 "the go-ahead's nonce {:?} was already spent on this box",
506 go.nonce
507 ));
508 }
509 let plan = Plan {
510 target_sectors: request.target_sectors_per_member,
511 drain_timeout: if request.drain_timeout_secs == 0 {
512 DEFAULT_DRAIN_TIMEOUT
513 } else {
514 Duration::from_secs(u64::from(request.drain_timeout_secs))
515 },
516 wait_timeout: if request.wait_members_timeout_secs == 0 {
517 DEFAULT_WAIT_MEMBERS_TIMEOUT
518 } else {
519 Duration::from_secs(u64::from(request.wait_members_timeout_secs))
520 },
521 signer: go.signer.clone(),
522 alone: {
523 let one_line = go.alone.replace(['\n', '\r'], " ");
524 let mut cut = one_line.trim().to_owned();
525 if cut.len() > ALONE_MAX {
526 let mut at = ALONE_MAX;
527 while !cut.is_char_boundary(at) {
528 at -= 1;
529 }
530 cut.truncate(at);
531 }
532 cut
533 },
534 };
535 {
536 let mut s = self.lock();
537 if s.phase.is_active() {
538 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));
539 }
540 if !s.serving {
541 return Err(format!("/data is not being served ({}); a growth cannot start from here — reboot, or re-attach the members and GrowResume", s.why));
542 }
543 let survey = self.world.survey();
544 if survey.verdict != "complete" {
545 return Err(format!(
546 "the data set is not a complete md set right now ({}); nothing to grow",
547 survey.verdict
548 ));
549 }
550 if plan.target_sectors > 0 && carries(&survey, plan.target_sectors) {
559 let described = survey.describe();
560 *s = State::idle(true);
561 s.signer = plan.signer.clone();
562 s.target_sectors = plan.target_sectors;
563 s.members = survey;
564 s.phase = Phase::Done;
565 s.since_unix_ms = now;
566 s.detail = format!(
567 "done: every member already carries the {} sectors asked ({described}); nothing was drained and /data never left",
568 plan.target_sectors
569 );
570 drop(s);
571 tracing::warn!(
572 target: TARGET,
573 event = "grow.already_met",
574 signer = %plan.signer,
575 nonce = %go.nonce,
576 target_sectors = plan.target_sectors,
577 "GrowStart for a size the members already carry: done at once, nothing drained or unmounted"
578 );
579 return Ok(self.status());
580 }
581 *s = State::idle(true);
582 s.signer = plan.signer.clone();
583 s.target_sectors = plan.target_sectors;
584 s.members = survey;
585 s.phase = Phase::Draining;
586 s.since_unix_ms = now;
587 s.detail = format!(
588 "draining: the gate is paused, waiting up to {} s for {} transfer(s) in flight",
589 plan.drain_timeout.as_secs(),
590 self.world.in_flight()
591 );
592 }
593 tracing::warn!(
594 target: TARGET,
595 event = "grow.start",
596 signer = %plan.signer,
597 nonce = %go.nonce,
598 target_sectors = plan.target_sectors,
599 drain_timeout_s = plan.drain_timeout.as_secs(),
600 wait_timeout_s = plan.wait_timeout.as_secs(),
601 "a runtime data-set growth begins"
602 );
603 let started = self.status();
608 let me = Arc::clone(self);
609 std::thread::Builder::new()
611 .name("gunnar-grow".into())
612 .spawn(move || me.run(plan))
613 .map_err(|e| format!("spawning the growth thread: {e}"))?;
614 Ok(started)
615 }
616
617 pub fn resume(&self, give_up: bool) -> Result<gunnar_proto::GrowStatusResponse, String> {
630 if self.lock().phase == Phase::Done {
631 tracing::info!(target: TARGET, event = "grow.resume.done", give_up, "GrowResume on a growth that is already done: nothing to resume");
632 return Ok(self.status());
633 }
634 {
635 let s = self.lock();
636 if matches!(s.phase, Phase::Growing | Phase::Mounting) {
645 drop(s);
646 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");
647 return Ok(self.status());
648 }
649 }
650 {
651 let mut s = self.lock();
652 if s.phase != Phase::WaitingMembers && s.phase != Phase::Unmounted {
655 return Err(format!(
656 "GrowResume applies while waiting-members; the growth is {} ({})",
657 s.phase.as_str(),
658 if s.phase == Phase::Idle {
659 "start one with GrowStart"
660 } else {
661 s.detail.as_str()
662 }
663 ));
664 }
665 s.resume = Some(if give_up {
666 Resume::GiveUp
667 } else {
668 Resume::LookAgain
669 });
670 }
671 tracing::info!(target: TARGET, event = "grow.resume", give_up, "GrowResume: looking at the members again");
672 self.wake.notify_all();
673 Ok(self.status())
674 }
675
676 pub fn status(&self) -> gunnar_proto::GrowStatusResponse {
680 let mut s = self.lock().clone();
681 if !s.phase.is_active() && self.world.engine_present() {
682 s.members = self.world.survey();
683 }
684 if s.phase == Phase::Draining {
685 s.in_flight = self.world.in_flight();
686 }
687 let journal = gunnar_grpc::grow_journal::Summary::read(&self.world.journal_path())
688 .map(|j| j.to_wire());
689 let detail = if s.detail.is_empty() {
690 match s.phase {
691 Phase::Idle if !self.world.engine_present() => {
692 "no runtime grow engine in this build; the set grows only at a restart"
693 .to_owned()
694 }
695 Phase::Idle => format!("idle; members: {}", s.members.describe()),
696 _ => String::new(),
697 }
698 } else {
699 s.detail.clone()
700 };
701 gunnar_proto::GrowStatusResponse {
702 phase: s.phase.as_str().to_owned(),
703 why: s.why.clone(),
704 epoch: s.epoch,
705 target_sectors_per_member: s.target_sectors,
706 members: s
707 .members
708 .members
709 .iter()
710 .map(|m| gunnar_proto::GrowMember {
711 index: m.index,
712 device: m.device.clone(),
713 partition: m.partition.clone(),
714 disk_bytes: m.disk_bytes,
715 set_bytes: m.set_bytes,
716 offered_bytes: m.offered_bytes,
717 larger: m.larger,
718 })
719 .collect(),
720 members_verdict: s.members.verdict.clone(),
721 journal,
722 engine_present: self.world.engine_present(),
723 hotplug: self.world.hotplug(),
724 serving: s.serving,
725 identity_in_ram: s.identity_in_ram,
726 in_flight: s.in_flight,
727 since_unix_ms: s.since_unix_ms,
728 server_unix_ms: gunnar_grpc::bus::now_unix_ms(),
729 detail,
730 go_ahead_signer: s.signer.clone(),
731 set_uuid: s.members.set_uuid.clone(),
732 total_bytes: if s.serving {
733 self.world.mounted_bytes()
734 } else {
735 0
736 },
737 }
738 }
739
740 fn run(self: Arc<Self>, plan: Plan) {
743 self.announce(Announcement::Draining {
746 kind: "growing",
747 reason: "the data set is being grown".into(),
748 });
749 self.world.pause_admission();
750 let started = Instant::now();
751 loop {
752 let in_flight = self.world.in_flight();
753 self.lock().in_flight = in_flight;
754 let starting = self.world.starting();
758 if in_flight == 0 && starting.is_none() {
759 break;
760 }
761 if started.elapsed() >= plan.drain_timeout {
762 self.world.unpause_admission();
763 let what = match &starting {
764 Some(why) => {
765 format!("{in_flight} transfer(s) still in flight and the box still {why}")
766 }
767 None => format!("{in_flight} transfer(s) still in flight"),
768 };
769 self.fail_untouched(format!(
770 "aborted: {what} after the {} s drain; nothing was changed — retry when the box is quieter, or raise drain_timeout_secs",
771 plan.drain_timeout.as_secs()
772 ));
773 return;
774 }
775 std::thread::sleep(DRAIN_POLL);
776 }
777 let flush_wait = if plan.alone.is_empty() {
806 plan.drain_timeout
807 } else {
808 plan.drain_timeout.min(ALONE_FLUSH_WAIT)
809 };
810 let verdict = match self.world.flush_replication(flush_wait) {
811 world::FlushVerdict::Behind {
812 pending,
813 needs_full_resync,
814 why,
815 } if !plan.alone.is_empty() => {
816 let said = format!("a standby is behind ({pending} pending, needs_full_resync {needs_full_resync}): {why}");
817 tracing::warn!(target: TARGET, event = "grow.alone", pending, %why, alone = %plan.alone, "the growth goes on ALONE: {said}");
818 world::FlushVerdict::NoStandby(format!("the growth goes on ALONE ({}) — {said}; the twin catches up from the log after the remount", plan.alone))
819 }
820 world::FlushVerdict::CouldNotLook(why) if !plan.alone.is_empty() => {
821 tracing::warn!(target: TARGET, event = "grow.alone", %why, alone = %plan.alone, "the growth goes on ALONE: the standby could not be looked at");
822 world::FlushVerdict::NoStandby(format!(
823 "the growth goes on ALONE ({}) — the standby could not be looked at: {why}",
824 plan.alone
825 ))
826 }
827 other => other,
828 };
829 if !plan.alone.is_empty() {
830 if let world::FlushVerdict::NoStandby(why) = &verdict {
831 self.lock().alone = why.clone();
832 }
833 }
834 match verdict {
835 world::FlushVerdict::CaughtUp {
836 standbys,
837 waited_ms,
838 } => {
839 tracing::info!(target: TARGET, event = "grow.flushed", standbys, waited_ms, "every standby has taken everything; the set may be unmounted");
840 self.lock().detail = format!("drained: nothing in flight and {standbys} standby(s) caught up after {waited_ms} ms");
841 }
842 world::FlushVerdict::NoStandby(why) => {
843 tracing::warn!(target: TARGET, event = "grow.no_standby", %why, "the set is being grown with no standby behind it");
847 self.lock().detail = format!("drained: nothing in flight, and {why}");
848 }
849 world::FlushVerdict::CouldNotLook(why) => {
850 self.world.unpause_admission();
856 self.fail_untouched(format!(
857 "aborted: the gate is shut and nothing is in flight, but whether the twin holds this set COULD NOT BE \
858 ESTABLISHED: {why} \u{2014} nothing was changed. A set unmounted now could only be pulled back from a twin, \
859 and nothing here says there is one that could."
860 ));
861 return;
862 }
863 world::FlushVerdict::Behind {
864 pending,
865 needs_full_resync,
866 why,
867 } => {
868 self.world.unpause_admission();
869 self.fail_untouched(format!(
870 "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."
871 ));
872 return;
873 }
874 }
875
876 let epoch = self.world.epoch_for(plan.target_sectors);
877 self.lock().epoch = epoch;
878 if let Err(e) = self.note(Step::Drain) {
879 self.world.unpause_admission();
880 self.fail_untouched(format!("aborted: {e}; nothing was changed"));
881 return;
882 }
883
884 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();
897 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)");
898 match self
899 .world
900 .suspend_identity("draining for the data-set growth")
901 {
902 Ok(facts) => {
903 self.lock().identity_in_ram = true;
904 tracing::info!(target: TARGET, event = "grow.identity_in_ram", documents = facts.documents, bytes = facts.bytes, "the control store is served from its RAM copy");
905 }
906 Err(e) => {
907 self.world.unpause_admission();
908 let mut why = format!("aborted: the identity table could not be copied into RAM ({e}); nothing was changed");
909 self.note_failing(Step::Abort, &mut why);
910 self.fail_untouched(why);
911 return;
912 }
913 }
914 if let Err(e) = self.world.close_stores() {
915 self.abort_after_suspend(format!("aborted: the stores would not close ({e})"));
916 return;
917 }
918 if let Err(e) = self.world.sync_and_unmount() {
919 self.abort_after_suspend(format!("aborted: {e}"));
920 return;
921 }
922 self.lock().serving = false;
923 if let Err(e) = self.note(Step::Unmount) {
924 self.recover_serve(format!("{e} after the unmount"));
925 return;
926 }
927 if let Err(e) = self.world.stop_array() {
928 self.recover_serve(format!("the array would not stop ({e})"));
929 return;
930 }
931 if let Err(e) = self.note(Step::Stop) {
932 self.recover_serve(format!("{e} after the array stopped"));
933 return;
934 }
935 self.enter(Phase::Unmounted, "unmounted: /data is off, the stores are closed, the array is stopped; the volumes are free");
936
937 if let Err(e) = self.note(Step::Wait) {
940 self.recover_serve(format!(
941 "the grow journal did not record WAIT ({e}); the members were not let go of"
942 ));
943 return;
944 }
945 self.enter(Phase::WaitingMembers, format!(
946 "waiting for every member to come back larger{}; the outside actor detaches, resizes and re-attaches the volumes; GrowResume looks again",
947 if plan.target_sectors > 0 { format!(" (≥ {} sectors each)", plan.target_sectors) } else { String::new() }
948 ));
949 let waited_from = Instant::now();
950 let ended = loop {
951 let timed_out = waited_from.elapsed() >= plan.wait_timeout;
954 if timed_out {
955 let mut s = self.lock();
956 if !s.wait_timed_out {
957 s.wait_timed_out = true;
958 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");
959 }
960 }
961 let _ = self.world.rescan();
962 let survey = self.world.survey();
963 let eligible = survey.eligible
971 && survey.target_sectors.is_some_and(|t| {
972 plan.target_sectors == 0 || target_reached(t, plan.target_sectors)
973 });
974 let same_set = survey.set_uuid.is_empty() || {
975 let s = self.lock();
976 s.members.set_uuid.is_empty() || s.members.set_uuid == survey.set_uuid
977 };
978 {
979 let mut s = self.lock();
980 let described = survey.describe();
981 s.members = survey;
982 s.detail = format!(
983 "waiting-members: {described}{}",
984 if !same_set {
985 " — NOT THE SAME SET (uuid differs); refusing to grow it"
986 } else if s.wait_timed_out {
987 " — the wait timeout passed; still waiting for GrowResume or the members"
988 } else {
989 ""
990 }
991 );
992 }
993 if eligible && same_set {
994 break Ended::Eligible;
995 }
996 if timed_out {
1007 let waited = waited_from.elapsed().as_secs();
1008 let described = self.lock().members.describe();
1009 let complete = self.lock().members.verdict == "complete";
1010 if complete && same_set {
1011 break Ended::Bound(format!(
1012 "{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",
1013 plan.wait_timeout.as_secs(),
1014 if plan.target_sectors > 0 { format!(" (≥ {} sectors each)", plan.target_sectors) } else { String::new() }
1015 ));
1016 }
1017 self.raise(
1018 UNMOUNTED_TOO_LONG,
1019 format!(
1020 "/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",
1021 plan.wait_timeout.as_secs(),
1022 if same_set { "a member is missing" } else { "the members are NOT the set that was unmounted" }
1023 ),
1024 false,
1025 );
1026 }
1027 let resume = self.lock().resume.take();
1028 match resume {
1029 Some(Resume::GiveUp) => break Ended::GaveUp,
1030 Some(Resume::LookAgain) => continue,
1031 None => {}
1032 }
1033 let poll = if timed_out {
1037 WAIT_POLL_AFTER_TIMEOUT
1038 } else {
1039 WAIT_POLL.min(
1040 plan.wait_timeout
1041 .saturating_sub(waited_from.elapsed())
1042 .max(Duration::from_millis(50)),
1043 )
1044 };
1045 let guard = self.lock();
1046 let (guard, _) = self
1047 .wake
1048 .wait_timeout(guard, poll)
1049 .unwrap_or_else(|p| p.into_inner());
1050 drop(guard);
1051 };
1052 if let Ended::Bound(why) = ended {
1053 self.raise(UNMOUNTED_TOO_LONG, why.clone(), true);
1054 self.recover_serve(why);
1055 return;
1056 }
1057 let give_up = matches!(ended, Ended::GaveUp);
1058 if let Err(e) = self.note(Step::Rescan) {
1059 tracing::warn!(target: TARGET, event = "grow.journal", %e, "RESCAN not recorded; going on (the disk is the truth)");
1060 }
1061
1062 self.enter(
1064 Phase::Growing,
1065 if give_up {
1066 "growing: the operator gave up waiting; assembling and mounting what is there"
1067 } else {
1068 "growing: every member is back and larger; the engine runs G1–G6"
1069 },
1070 );
1071 match self.world.grow_and_mount() {
1072 Ok(grown) => {
1073 self.lock().serving = true;
1074 self.lock().grown = Some(grown.clone());
1075 if grown.grew {
1076 if let Err(why) = self.reopen() {
1077 let why = format!("the data set grew ({}), but {why}", grown.describe());
1078 self.lock().why = why.clone();
1079 self.enter(Phase::Failed, format!("failed: {why}"));
1080 return;
1081 }
1082 let alone = std::mem::take(&mut self.lock().alone);
1083 self.enter(
1084 Phase::Done,
1085 if alone.is_empty() {
1086 format!("done: {}", grown.describe())
1087 } else {
1088 format!("done: {}; grown {alone}", grown.describe())
1089 },
1090 );
1091 tracing::warn!(target: TARGET, event = "grow.done", array_bytes = grown.array_bytes, "the data set grew and is being served");
1092 } else {
1093 let untouched =
1102 give_up && self.lock().members.members.iter().all(|m| !m.larger);
1103 let why = if give_up {
1104 format!(
1105 "gave up waiting{}; serving at the size the members share: {}",
1106 if untouched {
1107 " with no member back larger (nothing was touched)"
1108 } else {
1109 ""
1110 },
1111 grown.describe()
1112 )
1113 } else {
1114 format!("the engine did not grow the set; serving at the size the members share: {}", grown.describe())
1115 };
1116 let mut why = why;
1117 self.note_failing(if untouched { Step::Abort } else { Step::Fail }, &mut why);
1118 self.lock().why = why.clone();
1119 match self.reopen() {
1120 Ok(()) => self.enter(Phase::Failed, format!("failed (serving): {why}")),
1121 Err(more) => {
1122 let why = format!("{why}; and {more}");
1123 self.lock().why = why.clone();
1124 self.enter(Phase::Failed, format!("failed: {why}"));
1125 }
1126 }
1127 }
1128 }
1129 Err(e) => {
1130 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)");
1134 self.note_failing(Step::Fail, &mut why);
1135 self.lock().why = why.clone();
1136 self.raise("grow.fatal", why.clone(), true);
1137 self.enter(Phase::Failed, why);
1138 tracing::error!(target: TARGET, event = "grow.fatal", "{e}");
1139 }
1140 }
1141 }
1142
1143 fn fail_untouched(&self, why: String) {
1145 tracing::warn!(target: TARGET, event = "grow.aborted", %why, "the growth aborted with nothing changed");
1146 let mut s = self.lock();
1147 s.why = why.clone();
1148 s.serving = true;
1149 s.identity_in_ram = false;
1150 drop(s);
1151 self.enter(Phase::Failed, why);
1152 }
1153
1154 fn abort_after_suspend(&self, why: String) {
1160 let reopened = self.world.reopen_stores();
1161 match self.world.resume_identity() {
1162 Ok(r) => self.report_resume(&r),
1163 Err(e) => {
1164 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")
1165 }
1166 }
1167 let mut why = why;
1168 self.note_failing(Step::Abort, &mut why);
1169 match reopened {
1170 Ok(()) => {
1171 self.world.unpause_admission();
1172 self.fail_untouched(why);
1173 }
1174 Err(e) => {
1175 self.stores_stay_closed(format!("{why}; and the stores would not reopen ({e})"))
1176 }
1177 }
1178 }
1179
1180 fn stores_stay_closed(&self, why: String) {
1183 let why =
1184 format!("{why}: git stays paused over closed stores — restart the box to recover");
1185 {
1186 let mut s = self.lock();
1187 s.why = why.clone();
1188 s.serving = false;
1189 }
1190 self.raise("grow.reopen_failed", why.clone(), true);
1191 self.enter(Phase::Failed, why);
1192 }
1193
1194 fn recover_serve(&self, why: String) {
1198 tracing::warn!(target: TARGET, event = "grow.recover", %why, "recovering: mounting what is there and serving at the size it has");
1199 match self.world.grow_and_mount() {
1200 Ok(grown) => {
1201 self.lock().serving = true;
1202 self.lock().grown = Some(grown);
1203 let mut why = why;
1204 self.note_failing(Step::Fail, &mut why);
1205 self.lock().why = why.clone();
1206 match self.reopen() {
1207 Ok(()) => self.enter(
1208 Phase::Failed,
1209 format!("failed (serving at the old size): {why}"),
1210 ),
1211 Err(more) => {
1212 let why = format!("{why}; and {more}");
1213 self.lock().why = why.clone();
1214 self.enter(Phase::Failed, format!("failed: {why}"));
1215 }
1216 }
1217 }
1218 Err(e) => {
1219 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");
1220 self.note_failing(Step::Fail, &mut full);
1221 self.lock().why = full.clone();
1222 self.raise("grow.fatal", full.clone(), true);
1223 self.enter(Phase::Failed, full);
1224 tracing::error!(target: TARGET, event = "grow.fatal", "{e}");
1225 }
1226 }
1227 }
1228
1229 fn reopen(&self) -> Result<(), String> {
1236 self.enter(Phase::Mounting, "mounting: /data is back; the identity table returns to the stripe and the stores reopen");
1237 match self.world.resume_identity() {
1238 Ok(r) => {
1239 self.lock().identity_in_ram = false;
1240 self.report_resume(&r);
1241 }
1242 Err(e) => {
1243 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")
1244 }
1245 }
1246 if let Err(e) = self.world.reopen_stores() {
1247 let why = format!("the stores did not reopen ({e}): git stays paused over closed stores — restart the box to recover");
1248 self.lock().serving = false;
1249 self.raise("grow.reopen_failed", why.clone(), true);
1250 return Err(why);
1251 }
1252 let noted = self.note(Step::Reopen);
1253 self.world.unpause_admission();
1254 self.announce(Announcement::Ready);
1255 noted.map_err(|e| {
1256 tracing::error!(target: TARGET, event = "grow.journal", step = Step::Reopen.as_str(), %e, "the grow journal did not record REOPEN");
1257 format!("the grow journal did not record REOPEN ({e}); the stores are open and serving")
1258 })
1259 }
1260
1261 fn report_resume(&self, r: &world::ResumeFacts) {
1262 if r.drift.is_empty() {
1263 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");
1264 } else {
1265 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");
1266 let mut s = self.lock();
1267 s.why = format!(
1268 "{}{}identity drift on resume: {}",
1269 s.why,
1270 if s.why.is_empty() { "" } else { "; " },
1271 r.drift.join(", ")
1272 );
1273 }
1274 }
1275}
1276
1277#[cfg(test)]
1278pub(crate) mod tests;