LCOV - code coverage report
Current view: top level - safekeeper/src - json_ctrl.rs (source / functions) Coverage Total Hit
Test: 4f58e98c51285c7fa348e0b410c88a10caf68ad2.info Lines: 0.0 % 100 0
Test Date: 2025-01-07 20:58:07 Functions: 0.0 % 26 0

            Line data    Source code
       1              : //!
       2              : //! This module implements JSON_CTRL protocol, which allows exchange
       3              : //! JSON messages over psql for testing purposes.
       4              : //!
       5              : //! Currently supports AppendLogicalMessage, which is used for WAL
       6              : //! modifications in tests.
       7              : //!
       8              : 
       9              : use anyhow::Context;
      10              : use postgres_backend::QueryError;
      11              : use safekeeper_api::{ServerInfo, Term};
      12              : use serde::{Deserialize, Serialize};
      13              : use tokio::io::{AsyncRead, AsyncWrite};
      14              : use tracing::*;
      15              : 
      16              : use crate::handler::SafekeeperPostgresHandler;
      17              : use crate::safekeeper::{AcceptorProposerMessage, AppendResponse};
      18              : use crate::safekeeper::{
      19              :     AppendRequest, AppendRequestHeader, ProposerAcceptorMessage, ProposerElected,
      20              : };
      21              : use crate::safekeeper::{TermHistory, TermLsn};
      22              : use crate::state::TimelinePersistentState;
      23              : use crate::timeline::WalResidentTimeline;
      24              : use postgres_backend::PostgresBackend;
      25              : use postgres_ffi::encode_logical_message;
      26              : use postgres_ffi::WAL_SEGMENT_SIZE;
      27              : use pq_proto::{BeMessage, RowDescriptor, TEXT_OID};
      28              : use utils::lsn::Lsn;
      29              : 
      30            0 : #[derive(Serialize, Deserialize, Debug)]
      31              : pub struct AppendLogicalMessage {
      32              :     // prefix and message to build LogicalMessage
      33              :     pub lm_prefix: String,
      34              :     pub lm_message: String,
      35              : 
      36              :     // if true, commit_lsn will match flush_lsn after append
      37              :     pub set_commit_lsn: bool,
      38              : 
      39              :     // if true, ProposerElected will be sent before append
      40              :     pub send_proposer_elected: bool,
      41              : 
      42              :     // fields from AppendRequestHeader
      43              :     pub term: Term,
      44              :     #[serde(with = "utils::lsn::serde_as_u64")]
      45              :     pub epoch_start_lsn: Lsn,
      46              :     #[serde(with = "utils::lsn::serde_as_u64")]
      47              :     pub begin_lsn: Lsn,
      48              :     #[serde(with = "utils::lsn::serde_as_u64")]
      49              :     pub truncate_lsn: Lsn,
      50              :     pub pg_version: u32,
      51              : }
      52              : 
      53              : #[derive(Debug, Serialize)]
      54              : struct AppendResult {
      55              :     // safekeeper state after append
      56              :     state: TimelinePersistentState,
      57              :     // info about new record in the WAL
      58              :     inserted_wal: InsertedWAL,
      59              : }
      60              : 
      61              : /// Handles command to craft logical message WAL record with given
      62              : /// content, and then append it with specified term and lsn. This
      63              : /// function is used to test safekeepers in different scenarios.
      64            0 : pub async fn handle_json_ctrl<IO: AsyncRead + AsyncWrite + Unpin>(
      65            0 :     spg: &SafekeeperPostgresHandler,
      66            0 :     pgb: &mut PostgresBackend<IO>,
      67            0 :     append_request: &AppendLogicalMessage,
      68            0 : ) -> Result<(), QueryError> {
      69            0 :     info!("JSON_CTRL request: {append_request:?}");
      70              : 
      71              :     // need to init safekeeper state before AppendRequest
      72            0 :     let tli = prepare_safekeeper(spg, append_request.pg_version).await?;
      73              : 
      74              :     // if send_proposer_elected is true, we need to update local history
      75            0 :     if append_request.send_proposer_elected {
      76            0 :         send_proposer_elected(&tli, append_request.term, append_request.epoch_start_lsn).await?;
      77            0 :     }
      78              : 
      79            0 :     let inserted_wal = append_logical_message(&tli, append_request).await?;
      80            0 :     let response = AppendResult {
      81            0 :         state: tli.get_state().await.1,
      82            0 :         inserted_wal,
      83              :     };
      84            0 :     let response_data = serde_json::to_vec(&response)
      85            0 :         .with_context(|| format!("Response {response:?} is not a json array"))?;
      86              : 
      87            0 :     pgb.write_message_noflush(&BeMessage::RowDescription(&[RowDescriptor {
      88            0 :         name: b"json",
      89            0 :         typoid: TEXT_OID,
      90            0 :         typlen: -1,
      91            0 :         ..Default::default()
      92            0 :     }]))?
      93            0 :     .write_message_noflush(&BeMessage::DataRow(&[Some(&response_data)]))?
      94            0 :     .write_message_noflush(&BeMessage::CommandComplete(b"JSON_CTRL"))?;
      95            0 :     Ok(())
      96            0 : }
      97              : 
      98              : /// Prepare safekeeper to process append requests without crashes,
      99              : /// by sending ProposerGreeting with default server.wal_seg_size.
     100            0 : async fn prepare_safekeeper(
     101            0 :     spg: &SafekeeperPostgresHandler,
     102            0 :     pg_version: u32,
     103            0 : ) -> anyhow::Result<WalResidentTimeline> {
     104            0 :     let tli = spg
     105            0 :         .global_timelines
     106            0 :         .create(
     107            0 :             spg.ttid,
     108            0 :             ServerInfo {
     109            0 :                 pg_version,
     110            0 :                 wal_seg_size: WAL_SEGMENT_SIZE as u32,
     111            0 :                 system_id: 0,
     112            0 :             },
     113            0 :             Lsn::INVALID,
     114            0 :             Lsn::INVALID,
     115            0 :         )
     116            0 :         .await?;
     117              : 
     118            0 :     tli.wal_residence_guard().await
     119            0 : }
     120              : 
     121            0 : async fn send_proposer_elected(
     122            0 :     tli: &WalResidentTimeline,
     123            0 :     term: Term,
     124            0 :     lsn: Lsn,
     125            0 : ) -> anyhow::Result<()> {
     126              :     // add new term to existing history
     127            0 :     let history = tli.get_state().await.1.acceptor_state.term_history;
     128            0 :     let history = history.up_to(lsn.checked_sub(1u64).unwrap());
     129            0 :     let mut history_entries = history.0;
     130            0 :     history_entries.push(TermLsn { term, lsn });
     131            0 :     let history = TermHistory(history_entries);
     132            0 : 
     133            0 :     let proposer_elected_request = ProposerAcceptorMessage::Elected(ProposerElected {
     134            0 :         term,
     135            0 :         start_streaming_at: lsn,
     136            0 :         term_history: history,
     137            0 :         timeline_start_lsn: lsn,
     138            0 :     });
     139            0 : 
     140            0 :     tli.process_msg(&proposer_elected_request).await?;
     141            0 :     Ok(())
     142            0 : }
     143              : 
     144              : #[derive(Debug, Serialize)]
     145              : pub struct InsertedWAL {
     146              :     begin_lsn: Lsn,
     147              :     pub end_lsn: Lsn,
     148              :     append_response: AppendResponse,
     149              : }
     150              : 
     151              : /// Extend local WAL with new LogicalMessage record. To do that,
     152              : /// create AppendRequest with new WAL and pass it to safekeeper.
     153            0 : pub async fn append_logical_message(
     154            0 :     tli: &WalResidentTimeline,
     155            0 :     msg: &AppendLogicalMessage,
     156            0 : ) -> anyhow::Result<InsertedWAL> {
     157            0 :     let wal_data = encode_logical_message(&msg.lm_prefix, &msg.lm_message);
     158            0 :     let sk_state = tli.get_state().await.1;
     159              : 
     160            0 :     let begin_lsn = msg.begin_lsn;
     161            0 :     let end_lsn = begin_lsn + wal_data.len() as u64;
     162              : 
     163            0 :     let commit_lsn = if msg.set_commit_lsn {
     164            0 :         end_lsn
     165              :     } else {
     166            0 :         sk_state.commit_lsn
     167              :     };
     168              : 
     169            0 :     let append_request = ProposerAcceptorMessage::AppendRequest(AppendRequest {
     170            0 :         h: AppendRequestHeader {
     171            0 :             term: msg.term,
     172            0 :             term_start_lsn: begin_lsn,
     173            0 :             begin_lsn,
     174            0 :             end_lsn,
     175            0 :             commit_lsn,
     176            0 :             truncate_lsn: msg.truncate_lsn,
     177            0 :             proposer_uuid: [0u8; 16],
     178            0 :         },
     179            0 :         wal_data,
     180            0 :     });
     181              : 
     182            0 :     let response = tli.process_msg(&append_request).await?;
     183              : 
     184            0 :     let append_response = match response {
     185            0 :         Some(AcceptorProposerMessage::AppendResponse(resp)) => resp,
     186            0 :         _ => anyhow::bail!("not AppendResponse"),
     187              :     };
     188              : 
     189            0 :     Ok(InsertedWAL {
     190            0 :         begin_lsn,
     191            0 :         end_lsn,
     192            0 :         append_response,
     193            0 :     })
     194            0 : }
        

Generated by: LCOV version 2.1-beta