@@ -811,6 +811,12 @@ impl Manifest {
811811 /// Apply the destructive operation specified by the Garbage struct.
812812 #[ allow( clippy:: result_large_err) ]
813813 pub fn apply_garbage ( & self , mut garbage : Garbage ) -> Result < Self , Error > {
814+ if garbage. fragments_to_drop_start > garbage. fragments_to_drop_limit {
815+ return Err ( Error :: GarbageCollection ( format ! (
816+ "Garbage has start > limit: {:?} > {:?}" ,
817+ garbage. fragments_to_drop_start, garbage. fragments_to_drop_limit
818+ ) ) ) ;
819+ }
814820 let mut new = self . clone ( ) ;
815821 for to_drop in garbage. snapshots_to_drop . iter ( ) {
816822 if let Some ( index) = new. snapshots . iter ( ) . position ( |s| s == to_drop) {
@@ -834,8 +840,12 @@ impl Manifest {
834840 }
835841 new. collected += garbage. setsum_to_discard ;
836842 new. initial_offset = Some ( garbage. first_to_keep ) ;
837- if garbage. fragments_to_drop_start < garbage. fragments_to_drop_limit {
843+ if garbage. fragments_to_drop_start != FragmentSeqNo ( 0 )
844+ || garbage. fragments_to_drop_start < garbage. fragments_to_drop_limit
845+ {
838846 new. initial_seq_no = Some ( garbage. fragments_to_drop_limit ) ;
847+ } else {
848+ new. initial_seq_no = Some ( FragmentSeqNo ( 1 ) ) ;
839849 }
840850 new. scrub ( ) ?;
841851 Ok ( new)
@@ -1483,4 +1493,91 @@ mod tests {
14831493 LogPosition :: from_offset( 500 )
14841494 ) ;
14851495 }
1496+
1497+ #[ test]
1498+ fn apply_garbage_equal_nonzero_fragment_seq_nos ( ) {
1499+ let manifest = Manifest {
1500+ writer : "test_writer" . to_string ( ) ,
1501+ setsum : Setsum :: default ( ) ,
1502+ collected : Setsum :: default ( ) ,
1503+ acc_bytes : 0 ,
1504+ snapshots : vec ! [ ] ,
1505+ fragments : vec ! [ ] ,
1506+ initial_offset : None ,
1507+ initial_seq_no : None ,
1508+ } ;
1509+
1510+ let garbage = Garbage {
1511+ snapshots_to_drop : vec ! [ ] ,
1512+ snapshots_to_make : vec ! [ ] ,
1513+ fragments_to_drop_start : FragmentSeqNo ( 5 ) ,
1514+ fragments_to_drop_limit : FragmentSeqNo ( 5 ) ,
1515+ setsum_to_discard : Setsum :: default ( ) ,
1516+ first_to_keep : LogPosition :: from_offset ( 100 ) ,
1517+ snapshot_for_root : None ,
1518+ } ;
1519+
1520+ let result = manifest. apply_garbage ( garbage) . unwrap ( ) ;
1521+
1522+ // When fragments_to_drop_start == fragments_to_drop_limit and both are non-zero,
1523+ // initial_seq_no should be set to the limit value
1524+ assert_eq ! ( result. initial_seq_no, Some ( FragmentSeqNo ( 5 ) ) ) ;
1525+ assert_eq ! ( result. initial_offset, Some ( LogPosition :: from_offset( 100 ) ) ) ;
1526+ }
1527+
1528+ #[ test]
1529+ fn apply_garbage_validates_fragment_drop_range ( ) {
1530+ use crate :: gc:: Garbage ;
1531+
1532+ let manifest = Manifest :: new_empty ( "test" ) ;
1533+
1534+ // Test case: fragments_to_drop_start > fragments_to_drop_limit should fail
1535+ let invalid_garbage = Garbage {
1536+ snapshots_to_drop : vec ! [ ] ,
1537+ snapshots_to_make : vec ! [ ] ,
1538+ snapshot_for_root : None ,
1539+ fragments_to_drop_start : FragmentSeqNo ( 10 ) ,
1540+ fragments_to_drop_limit : FragmentSeqNo ( 5 ) ,
1541+ setsum_to_discard : Setsum :: default ( ) ,
1542+ first_to_keep : LogPosition :: from_offset ( 1 ) ,
1543+ } ;
1544+
1545+ let result = manifest. apply_garbage ( invalid_garbage) ;
1546+ assert ! ( result. is_err( ) ) ;
1547+
1548+ if let Err ( crate :: Error :: GarbageCollection ( msg) ) = result {
1549+ assert ! ( msg. contains( "Garbage has start > limit" ) ) ;
1550+ assert ! ( msg. contains( "FragmentSeqNo(10) > FragmentSeqNo(5)" ) ) ;
1551+ } else {
1552+ panic ! ( "Expected GarbageCollection error, got {:?}" , result) ;
1553+ }
1554+
1555+ // Test case: fragments_to_drop_start == fragments_to_drop_limit should succeed
1556+ let valid_garbage_equal = Garbage {
1557+ snapshots_to_drop : vec ! [ ] ,
1558+ snapshots_to_make : vec ! [ ] ,
1559+ snapshot_for_root : None ,
1560+ fragments_to_drop_start : FragmentSeqNo ( 5 ) ,
1561+ fragments_to_drop_limit : FragmentSeqNo ( 5 ) ,
1562+ setsum_to_discard : Setsum :: default ( ) ,
1563+ first_to_keep : LogPosition :: from_offset ( 1 ) ,
1564+ } ;
1565+
1566+ let result = manifest. apply_garbage ( valid_garbage_equal) ;
1567+ assert ! ( result. is_ok( ) ) ;
1568+
1569+ // Test case: fragments_to_drop_start < fragments_to_drop_limit should succeed
1570+ let valid_garbage_less = Garbage {
1571+ snapshots_to_drop : vec ! [ ] ,
1572+ snapshots_to_make : vec ! [ ] ,
1573+ snapshot_for_root : None ,
1574+ fragments_to_drop_start : FragmentSeqNo ( 1 ) ,
1575+ fragments_to_drop_limit : FragmentSeqNo ( 5 ) ,
1576+ setsum_to_discard : Setsum :: default ( ) ,
1577+ first_to_keep : LogPosition :: from_offset ( 1 ) ,
1578+ } ;
1579+
1580+ let result = manifest. apply_garbage ( valid_garbage_less) ;
1581+ assert ! ( result. is_ok( ) ) ;
1582+ }
14861583}
0 commit comments