turbopack_dev_server/
lib.rs

1#![feature(min_specialization)]
2#![feature(str_split_remainder)]
3#![feature(arbitrary_self_types)]
4#![feature(arbitrary_self_types_pointers)]
5
6pub mod html;
7mod http;
8pub mod introspect;
9mod invalidation;
10pub mod source;
11pub mod update;
12
13use std::{
14    collections::VecDeque,
15    future::Future,
16    net::{SocketAddr, TcpListener},
17    pin::Pin,
18    sync::Arc,
19    time::{Duration, Instant},
20};
21
22use anyhow::{Context, Result};
23use hyper::{
24    Request, Response, Server,
25    server::{Builder, conn::AddrIncoming},
26    service::{make_service_fn, service_fn},
27};
28use parking_lot::Mutex;
29use socket2::{Domain, Protocol, Socket, Type};
30use tokio::task::JoinHandle;
31use tracing::{Instrument, Level, Span, event, info_span};
32use turbo_tasks::{
33    Completion, Effects, NonLocalValue, OperationVc, PrettyPrintError, ResolvedVc, TurboTasksApi,
34    Vc, read_strongly_consistent_and_apply_effects, run_once_with_reason, take_effects,
35    trace::TraceRawVcs, util::FormatDuration,
36};
37use turbopack_core::issue::{IssueReporter, IssueSeverity, handle_issues};
38
39use self::{source::ContentSource, update::UpdateServer};
40use crate::{
41    invalidation::{ServerRequest, ServerRequestSideEffects},
42    source::ContentSourceSideEffect,
43};
44
45pub trait SourceProvider: Send + Clone + 'static {
46    /// must call a turbo-tasks function internally
47    fn get_source(&self) -> OperationVc<Box<dyn ContentSource>>;
48}
49
50impl<T> SourceProvider for T
51where
52    T: Fn() -> OperationVc<Box<dyn ContentSource>> + Send + Clone + 'static,
53{
54    fn get_source(&self) -> OperationVc<Box<dyn ContentSource>> {
55        self()
56    }
57}
58
59#[turbo_tasks::value(serialization = "skip")]
60struct ContentSourceWithIssues {
61    source_op: OperationVc<Box<dyn ContentSource>>,
62    effects: Effects,
63}
64
65#[turbo_tasks::function(operation, root)]
66async fn get_source_with_issues_operation(
67    source_op: OperationVc<Box<dyn ContentSource>>,
68) -> Result<Vc<ContentSourceWithIssues>> {
69    let _ = source_op.resolve().strongly_consistent().await?;
70    let effects = take_effects(source_op).await?;
71    Ok(ContentSourceWithIssues { source_op, effects }.cell())
72}
73
74/// Applies all collected [`ContentSourceSideEffect`]s. The individual `apply()` reads happen
75/// *inside* this task (where eventually-consistent reads are legal); the caller reads the result
76/// strongly consistently so the work is finished before the response is observed.
77#[turbo_tasks::function(operation, root)]
78async fn apply_side_effects_operation(
79    side_effects: Vec<ResolvedVc<Box<dyn ContentSourceSideEffect>>>,
80) -> Result<Vc<Completion>> {
81    for side_effect in &side_effects {
82        side_effect.apply().await?;
83    }
84    Ok(Completion::new())
85}
86
87#[derive(TraceRawVcs, Debug, NonLocalValue)]
88pub struct DevServerBuilder {
89    #[turbo_tasks(trace_ignore)]
90    pub addr: SocketAddr,
91    #[turbo_tasks(trace_ignore)]
92    server: Builder<AddrIncoming>,
93}
94
95#[derive(TraceRawVcs, NonLocalValue)]
96pub struct DevServer {
97    #[turbo_tasks(trace_ignore)]
98    pub addr: SocketAddr,
99    #[turbo_tasks(trace_ignore)]
100    pub future: Pin<Box<dyn Future<Output = Result<()>> + Send + 'static>>,
101}
102
103impl DevServer {
104    pub fn listen(addr: SocketAddr) -> Result<DevServerBuilder, anyhow::Error> {
105        // This is annoying. The hyper::Server doesn't allow us to know which port was
106        // bound (until we build it with a request handler) when using the standard
107        // `server::try_bind` approach. This is important when binding the `0` port,
108        // because the OS will remap that to an actual free port, and we need to know
109        // that port before we build the request handler. So we need to construct a
110        // real TCP listener, see if it bound, and get its bound address.
111        let socket = Socket::new(Domain::for_address(addr), Type::STREAM, Some(Protocol::TCP))
112            .context("unable to create socket")?;
113        // Allow the socket to be reused immediately after closing. This ensures that
114        // the dev server can be restarted on the same address without a buffer time for
115        // the OS to release the socket.
116        // https://docs.microsoft.com/en-us/windows/win32/winsock/using-so-reuseaddr-and-so-exclusiveaddruse
117        #[cfg(not(windows))]
118        let _ = socket.set_reuse_address(true);
119        if matches!(addr, SocketAddr::V6(_)) {
120            // When possible bind to v4 and v6, otherwise ignore the error
121            let _ = socket.set_only_v6(false);
122        }
123        let sock_addr = addr.into();
124        socket
125            .bind(&sock_addr)
126            .context("not able to bind address")?;
127        socket.listen(128).context("not able to listen on socket")?;
128
129        let listener: TcpListener = socket.into();
130        let addr = listener
131            .local_addr()
132            .context("not able to get bound address")?;
133        let server = Server::from_tcp(listener).context("Not able to start server")?;
134        Ok(DevServerBuilder { addr, server })
135    }
136}
137
138impl DevServerBuilder {
139    pub fn serve(
140        self,
141        turbo_tasks: Arc<dyn TurboTasksApi>,
142        source_provider: impl SourceProvider + NonLocalValue + TraceRawVcs + Sync,
143        get_issue_reporter: Arc<dyn Fn() -> Vc<Box<dyn IssueReporter>> + Send + Sync>,
144    ) -> DevServer {
145        let ongoing_side_effects = Arc::new(Mutex::new(VecDeque::<
146            Arc<tokio::sync::Mutex<Option<JoinHandle<Result<()>>>>>,
147        >::with_capacity(16)));
148        let make_svc = make_service_fn(move |_| {
149            let tt = turbo_tasks.clone();
150            let source_provider = source_provider.clone();
151            let get_issue_reporter = get_issue_reporter.clone();
152            let ongoing_side_effects = ongoing_side_effects.clone();
153            async move {
154                let handler = move |request: Request<hyper::Body>| {
155                    let request_span = info_span!(parent: None, "request", name = ?request.uri());
156                    let start = Instant::now();
157                    let tt = tt.clone();
158                    let get_issue_reporter = get_issue_reporter.clone();
159                    let ongoing_side_effects = ongoing_side_effects.clone();
160                    let source_provider = source_provider.clone();
161                    let future = async move {
162                        event!(parent: Span::current(), Level::DEBUG, "request start");
163                        // Wait until all ongoing side effects are completed
164                        // We only need to wait for the ongoing side effects that were started
165                        // before this request. Later added side effects are not relevant for this.
166                        let current_ongoing_side_effects = {
167                            // Cleanup the ongoing_side_effects list
168                            let mut guard = ongoing_side_effects.lock();
169                            while let Some(front) = guard.front() {
170                                let Ok(front_guard) = front.try_lock() else {
171                                    break;
172                                };
173                                if front_guard.is_some() {
174                                    break;
175                                }
176                                drop(front_guard);
177                                guard.pop_front();
178                            }
179                            // Get a clone of the remaining list
180                            (*guard).clone()
181                        };
182                        // Wait for the side effects to complete
183                        for side_effect_mutex in current_ongoing_side_effects {
184                            let mut guard = side_effect_mutex.lock().await;
185                            if let Some(join_handle) = guard.take() {
186                                join_handle.await??;
187                            }
188                            drop(guard);
189                        }
190                        let reason = ServerRequest {
191                            method: request.method().clone(),
192                            uri: request.uri().clone(),
193                        };
194                        let side_effects_reason = ServerRequestSideEffects {
195                            method: request.method().clone(),
196                            uri: request.uri().clone(),
197                        };
198                        run_once_with_reason(tt.clone(), reason, async move {
199                            // TODO: `get_issue_reporter` should be an `OperationVc`, as there's a
200                            // risk it could be a task-local Vc, which is not safe for us to await.
201                            let issue_reporter = get_issue_reporter();
202
203                            if hyper_tungstenite::is_upgrade_request(&request) {
204                                let uri = request.uri();
205                                let path = uri.path();
206
207                                if path == "/turbopack-hmr" {
208                                    let (response, websocket) =
209                                        hyper_tungstenite::upgrade(request, None)?;
210                                    let update_server =
211                                        UpdateServer::new(source_provider, issue_reporter);
212                                    update_server.run(&*tt, websocket);
213                                    return Ok(response);
214                                }
215
216                                println!("[404] {path} (WebSocket)");
217                                if path == "/_next/hmr" {
218                                    // Special-case requests to hmr as these are made by
219                                    // Next.js clients built
220                                    // without turbopack, which may be making requests in
221                                    // development.
222                                    println!(
223                                        "A non-turbopack next.js client is trying to connect."
224                                    );
225                                    println!(
226                                        "Make sure to reload/close any browser window which has \
227                                         been opened without --turbo."
228                                    );
229                                }
230
231                                return Ok(Response::builder()
232                                    .status(404)
233                                    .body(hyper::Body::empty())?);
234                            }
235
236                            let uri = request.uri();
237                            let path = uri.path().to_string();
238                            let source_with_issues_op =
239                                get_source_with_issues_operation(source_provider.get_source());
240                            let read = read_strongly_consistent_and_apply_effects(
241                                source_with_issues_op,
242                                |v| &v.effects,
243                            )
244                            .await?;
245                            let ContentSourceWithIssues { source_op, .. } = &*read;
246                            handle_issues(
247                                source_with_issues_op,
248                                issue_reporter,
249                                IssueSeverity::Fatal,
250                                Some(&path),
251                                Some("get source"),
252                            )
253                            .await?;
254                            let (response, side_effects) =
255                                http::process_request_with_content_source(
256                                    // HACK: pass `source` here (instead of `resolved_source`
257                                    // because the underlying API wants to do it's own
258                                    // `.resolve().strongly_consistent()` call.
259                                    //
260                                    // It's unlikely (the calls happen one-after-another), but this
261                                    // could cause inconsistency between the reported issues and
262                                    // the generated HTTP response.
263                                    *source_op,
264                                    request,
265                                    issue_reporter,
266                                )
267                                .await?;
268                            let status = response.status().as_u16();
269                            let is_error = response.status().is_client_error()
270                                || response.status().is_server_error();
271                            let elapsed = start.elapsed();
272                            if is_error
273                                || (cfg!(feature = "log_request_stats")
274                                    && elapsed > Duration::from_secs(1))
275                            {
276                                println!(
277                                    "[{status}] {path} ({duration})",
278                                    duration = FormatDuration(elapsed)
279                                );
280                            }
281                            if !side_effects.is_empty() {
282                                let side_effects: Vec<_> = side_effects.into_iter().collect();
283                                let join_handle = tokio::spawn(run_once_with_reason(
284                                    tt.clone(),
285                                    side_effects_reason,
286                                    async move {
287                                        // Apply the side effects inside a dedicated `operation`
288                                        // task and read its result strongly consistently, so this
289                                        // top-level task performs no eventually-consistent read.
290                                        apply_side_effects_operation(side_effects)
291                                            .read_strongly_consistent()
292                                            .await?;
293                                        Ok(())
294                                    },
295                                ));
296                                ongoing_side_effects.lock().push_back(Arc::new(
297                                    tokio::sync::Mutex::new(Some(join_handle)),
298                                ));
299                            }
300                            Ok(response)
301                        })
302                        .await
303                    };
304                    async move {
305                        match future.await {
306                            Ok(r) => Ok::<_, hyper::http::Error>(r),
307                            Err(e) => {
308                                println!(
309                                    "[500] error ({}): {}",
310                                    FormatDuration(start.elapsed()),
311                                    PrettyPrintError(&e),
312                                );
313                                Ok(Response::builder()
314                                    .status(500)
315                                    .body(hyper::Body::from(format!("{}", PrettyPrintError(&e))))?)
316                            }
317                        }
318                    }
319                    .instrument(request_span)
320                };
321                anyhow::Ok(service_fn(handler))
322            }
323        });
324        let server = self.server.serve(make_svc);
325
326        DevServer {
327            addr: self.addr,
328            future: Box::pin(async move {
329                server.await?;
330                Ok(())
331            }),
332        }
333    }
334}
turbopack_dev_server/lib.rs

turbopack_dev_server/
lib.rs
